forked from hyperledger-labs/yui-relayer
-
Notifications
You must be signed in to change notification settings - Fork 0
/
relayMsgs.go
113 lines (92 loc) · 2.65 KB
/
relayMsgs.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
package core
import (
sdk "github.com/cosmos/cosmos-sdk/types"
"github.com/gogo/protobuf/proto"
)
// RelayMsgs contains the msgs that need to be sent to both a src and dst chain
// after a given relay round. MaxTxSize and MaxMsgLength are ignored if they are
// set to zero.
type RelayMsgs struct {
Src []sdk.Msg `json:"src"`
Dst []sdk.Msg `json:"dst"`
MaxTxSize uint64 `json:"max_tx_size"` // maximum permitted size of the msgs in a bundled relay transaction
MaxMsgLength uint64 `json:"max_msg_length"` // maximum amount of messages in a bundled relay transaction
Last bool `json:"last"`
Succeeded bool `json:"success"`
}
// NewRelayMsgs returns an initialized version of relay messages
func NewRelayMsgs() *RelayMsgs {
return &RelayMsgs{Src: []sdk.Msg{}, Dst: []sdk.Msg{}, Last: false, Succeeded: false}
}
// Ready returns true if there are messages to relay
func (r *RelayMsgs) Ready() bool {
if r == nil {
return false
}
if len(r.Src) == 0 && len(r.Dst) == 0 {
return false
}
return true
}
// Success returns the success var
func (r *RelayMsgs) Success() bool {
return r.Succeeded
}
func (r *RelayMsgs) IsMaxTx(msgLen, txSize uint64) bool {
return (r.MaxMsgLength != 0 && msgLen > r.MaxMsgLength) ||
(r.MaxTxSize != 0 && txSize > r.MaxTxSize)
}
// Send sends the messages with appropriate output
// TODO: Parallelize? Maybe?
func (r *RelayMsgs) Send(src, dst ChainI) {
//nolint:prealloc // can not be pre allocated
var (
msgLen, txSize uint64
msgs []sdk.Msg
)
r.Succeeded = true
// submit batches of relay transactions
for _, msg := range r.Src {
bz, err := proto.Marshal(msg)
if err != nil {
panic(err)
}
msgLen++
txSize += uint64(len(bz))
if r.IsMaxTx(msgLen, txSize) {
// Submit the transactions to src chain and update its status
r.Succeeded = r.Succeeded && src.Send(msgs)
// clear the current batch and reset variables
msgLen, txSize = 1, uint64(len(bz))
msgs = []sdk.Msg{}
}
msgs = append(msgs, msg)
}
// submit leftover msgs
if len(msgs) > 0 && !src.Send(msgs) {
r.Succeeded = false
}
// reset variables
msgLen, txSize = 0, 0
msgs = []sdk.Msg{}
for _, msg := range r.Dst {
bz, err := proto.Marshal(msg)
if err != nil {
panic(err)
}
msgLen++
txSize += uint64(len(bz))
if r.IsMaxTx(msgLen, txSize) {
// Submit the transaction to dst chain and update its status
r.Succeeded = r.Succeeded && dst.Send(msgs)
// clear the current batch and reset variables
msgLen, txSize = 1, uint64(len(bz))
msgs = []sdk.Msg{}
}
msgs = append(msgs, msg)
}
// submit leftover msgs
if len(msgs) > 0 && !dst.Send(msgs) {
r.Succeeded = false
}
}