-
Notifications
You must be signed in to change notification settings - Fork 204
/
publish.go
123 lines (100 loc) · 2.63 KB
/
publish.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
116
117
118
119
120
121
122
123
package events
import (
"context"
"golang.org/x/sync/errgroup"
sdk "github.com/cosmos/cosmos-sdk/types"
abci "github.com/tendermint/tendermint/abci/types"
tmclient "github.com/tendermint/tendermint/rpc/client"
ctypes "github.com/tendermint/tendermint/rpc/core/types"
tmtmtypes "github.com/tendermint/tendermint/types"
atypes "github.com/akash-network/akash-api/go/node/audit/v1beta3"
dtypes "github.com/akash-network/akash-api/go/node/deployment/v1beta3"
mtypes "github.com/akash-network/akash-api/go/node/market/v1beta4"
ptypes "github.com/akash-network/akash-api/go/node/provider/v1beta3"
"github.com/akash-network/akash-api/go/sdkutil"
"github.com/akash-network/node/pubsub"
)
// Publish events using tm buses to clients. Waits on context
// shutdown signals to exit.
func Publish(ctx context.Context, tmbus tmclient.EventsClient, name string, bus pubsub.Bus) (err error) {
const (
queuesz = 100
)
var (
txname = name + "-tx"
blkname = name + "-blk"
)
txch, err := tmbus.Subscribe(ctx, txname, txQuery().String(), queuesz)
if err != nil {
return err
}
defer func() {
err = tmbus.UnsubscribeAll(ctx, txname)
}()
blkch, err := tmbus.Subscribe(ctx, blkname, blkQuery().String(), queuesz)
if err != nil {
return err
}
defer func() {
err = tmbus.UnsubscribeAll(ctx, blkname)
}()
g, ctx := errgroup.WithContext(ctx)
g.Go(func() error {
return publishEvents(ctx, txch, bus)
})
g.Go(func() error {
return publishEvents(ctx, blkch, bus)
})
return g.Wait()
}
func publishEvents(ctx context.Context, ch <-chan ctypes.ResultEvent, bus pubsub.Bus) error {
var err error
loop:
for {
select {
case <-ctx.Done():
break loop
case ed := <-ch:
switch evt := ed.Data.(type) {
case tmtmtypes.EventDataTx:
if !evt.Result.IsOK() {
continue
}
processEvents(bus, evt.Result.GetEvents())
case tmtmtypes.EventDataNewBlockHeader:
processEvents(bus, evt.ResultEndBlock.GetEvents())
}
}
}
return err
}
func processEvents(bus pubsub.Bus, events []abci.Event) {
for _, ev := range events {
if mev, ok := processEvent(ev); ok {
if err := bus.Publish(mev); err != nil {
bus.Close()
return
}
continue
}
}
}
func processEvent(bev abci.Event) (interface{}, bool) {
ev, err := sdkutil.ParseEvent(sdk.StringifyEvent(bev))
if err != nil {
return nil, false
}
if mev, err := dtypes.ParseEvent(ev); err == nil {
return mev, true
}
if mev, err := mtypes.ParseEvent(ev); err == nil {
return mev, true
}
if mev, err := ptypes.ParseEvent(ev); err == nil {
return mev, true
}
if mev, err := atypes.ParseEvent(ev); err == nil {
return mev, true
}
return nil, false
}