-
Notifications
You must be signed in to change notification settings - Fork 38
/
handlers.go
107 lines (87 loc) · 3 KB
/
handlers.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
package neofs
import (
"encoding/hex"
"github.com/nspcc-dev/neo-go/pkg/util/slice"
"github.com/nspcc-dev/neofs-node/pkg/morph/event"
neofsEvent "github.com/nspcc-dev/neofs-node/pkg/morph/event/neofs"
"go.uber.org/zap"
)
func (np *Processor) handleDeposit(ev event.Event) {
deposit := ev.(neofsEvent.Deposit)
np.log.Info("notification",
zap.String("type", "deposit"),
zap.String("id", hex.EncodeToString(slice.CopyReverse(deposit.ID()))))
// send event to the worker pool
err := np.pool.Submit(func() { np.processDeposit(&deposit) })
if err != nil {
// there system can be moved into controlled degradation stage
np.log.Warn("neofs processor worker pool drained",
zap.Int("capacity", np.pool.Cap()))
}
}
func (np *Processor) handleWithdraw(ev event.Event) {
withdraw := ev.(neofsEvent.Withdraw)
np.log.Info("notification",
zap.String("type", "withdraw"),
zap.String("id", hex.EncodeToString(slice.CopyReverse(withdraw.ID()))))
// send event to the worker pool
err := np.pool.Submit(func() { np.processWithdraw(&withdraw) })
if err != nil {
// there system can be moved into controlled degradation stage
np.log.Warn("neofs processor worker pool drained",
zap.Int("capacity", np.pool.Cap()))
}
}
func (np *Processor) handleCheque(ev event.Event) {
cheque := ev.(neofsEvent.Cheque)
np.log.Info("notification",
zap.String("type", "cheque"),
zap.String("id", hex.EncodeToString(cheque.ID())))
// send event to the worker pool
err := np.pool.Submit(func() { np.processCheque(&cheque) })
if err != nil {
// there system can be moved into controlled degradation stage
np.log.Warn("neofs processor worker pool drained",
zap.Int("capacity", np.pool.Cap()))
}
}
func (np *Processor) handleConfig(ev event.Event) {
cfg := ev.(neofsEvent.Config)
np.log.Info("notification",
zap.String("type", "set config"),
zap.String("key", hex.EncodeToString(cfg.Key())),
zap.String("value", hex.EncodeToString(cfg.Value())))
// send event to the worker pool
err := np.pool.Submit(func() { np.processConfig(&cfg) })
if err != nil {
// there system can be moved into controlled degradation stage
np.log.Warn("neofs processor worker pool drained",
zap.Int("capacity", np.pool.Cap()))
}
}
func (np *Processor) handleBind(ev event.Event) {
e := ev.(neofsEvent.Bind)
np.log.Info("notification",
zap.String("type", "bind"),
)
// send event to the worker pool
err := np.pool.Submit(func() { np.processBind(e) })
if err != nil {
// there system can be moved into controlled degradation stage
np.log.Warn("neofs processor worker pool drained",
zap.Int("capacity", np.pool.Cap()))
}
}
func (np *Processor) handleUnbind(ev event.Event) {
e := ev.(neofsEvent.Unbind)
np.log.Info("notification",
zap.String("type", "unbind"),
)
// send event to the worker pool
err := np.pool.Submit(func() { np.processBind(e) })
if err != nil {
// there system can be moved into controlled degradation stage
np.log.Warn("neofs processor worker pool drained",
zap.Int("capacity", np.pool.Cap()))
}
}