/
group_processors.go
115 lines (100 loc) · 3.17 KB
/
group_processors.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
package ingest
import (
"context"
"fmt"
"time"
"github.com/lantah/go/ingest"
"github.com/lantah/go/services/orbitr/internal/ingest/processors"
"github.com/lantah/go/support/errors"
)
type processorsRunDurations map[string]time.Duration
func (d processorsRunDurations) AddRunDuration(name string, startTime time.Time) {
d[name] += time.Since(startTime)
}
type groupChangeProcessors struct {
processors []orbitrChangeProcessor
processorsRunDurations
}
func newGroupChangeProcessors(processors []orbitrChangeProcessor) *groupChangeProcessors {
return &groupChangeProcessors{
processors: processors,
processorsRunDurations: make(map[string]time.Duration),
}
}
func (g groupChangeProcessors) ProcessChange(ctx context.Context, change ingest.Change) error {
for _, p := range g.processors {
startTime := time.Now()
if err := p.ProcessChange(ctx, change); err != nil {
return errors.Wrapf(err, "error in %T.ProcessChange", p)
}
g.AddRunDuration(fmt.Sprintf("%T", p), startTime)
}
return nil
}
func (g groupChangeProcessors) Commit(ctx context.Context) error {
for _, p := range g.processors {
startTime := time.Now()
if err := p.Commit(ctx); err != nil {
return errors.Wrapf(err, "error in %T.Commit", p)
}
g.AddRunDuration(fmt.Sprintf("%T", p), startTime)
}
return nil
}
type groupTransactionProcessors struct {
processors []orbitrTransactionProcessor
processorsRunDurations
}
func newGroupTransactionProcessors(processors []orbitrTransactionProcessor) *groupTransactionProcessors {
return &groupTransactionProcessors{
processors: processors,
processorsRunDurations: make(map[string]time.Duration),
}
}
func (g groupTransactionProcessors) ProcessTransaction(ctx context.Context, tx ingest.LedgerTransaction) error {
for _, p := range g.processors {
startTime := time.Now()
if err := p.ProcessTransaction(ctx, tx); err != nil {
return errors.Wrapf(err, "error in %T.ProcessTransaction", p)
}
g.AddRunDuration(fmt.Sprintf("%T", p), startTime)
}
return nil
}
func (g groupTransactionProcessors) Commit(ctx context.Context) error {
for _, p := range g.processors {
startTime := time.Now()
if err := p.Commit(ctx); err != nil {
return errors.Wrapf(err, "error in %T.Commit", p)
}
g.AddRunDuration(fmt.Sprintf("%T", p), startTime)
}
return nil
}
type groupTransactionFilterers struct {
filterers []processors.LedgerTransactionFilterer
processorsRunDurations
droppedTransactions int64
}
func newGroupTransactionFilterers(filterers []processors.LedgerTransactionFilterer) *groupTransactionFilterers {
return &groupTransactionFilterers{
filterers: filterers,
processorsRunDurations: make(map[string]time.Duration),
}
}
func (g *groupTransactionFilterers) FilterTransaction(ctx context.Context, tx ingest.LedgerTransaction) (bool, error) {
for _, f := range g.filterers {
startTime := time.Now()
include, err := f.FilterTransaction(ctx, tx)
if err != nil {
return false, errors.Wrapf(err, "error in %T.FilterTransaction", f)
}
g.AddRunDuration(fmt.Sprintf("%T", f), startTime)
if !include {
// filter out, we can return early
g.droppedTransactions++
return false, nil
}
}
return true, nil
}