forked from stellar/go
-
Notifications
You must be signed in to change notification settings - Fork 0
/
group_processors.go
84 lines (72 loc) · 2.23 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
package ingest
import (
"context"
"fmt"
"time"
"github.com/stellar/go/ingest"
"github.com/stellar/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 []horizonChangeProcessor
processorsRunDurations
}
func newGroupChangeProcessors(processors []horizonChangeProcessor) *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 []horizonTransactionProcessor
processorsRunDurations
}
func newGroupTransactionProcessors(processors []horizonTransactionProcessor) *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
}