-
Notifications
You must be signed in to change notification settings - Fork 662
/
eventsocket.go
149 lines (123 loc) · 3.76 KB
/
eventsocket.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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
// (c) 2020, Ava Labs, Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package ipcs
import (
"errors"
"os"
"syscall"
"github.com/ava-labs/avalanchego/ids"
"github.com/ava-labs/avalanchego/ipcs/socket"
"github.com/ava-labs/avalanchego/snow"
"github.com/ava-labs/avalanchego/snow/triggers"
"github.com/ava-labs/avalanchego/utils/logging"
"github.com/ava-labs/avalanchego/utils/wrappers"
)
// EventSockets is a set of named eventSockets
type EventSockets struct {
consensusSocket *eventSocket
decisionsSocket *eventSocket
}
// newEventSockets creates a *ChainIPCs with both consensus and decisions IPCs
func newEventSockets(ctx context, chainID ids.ID, consensusEvents *triggers.EventDispatcher, decisionEvents *triggers.EventDispatcher) (*EventSockets, error) {
consensusIPC, err := newEventIPCSocket(ctx, chainID, ipcConsensusIdentifier, consensusEvents)
if err != nil {
return nil, err
}
decisionsIPC, err := newEventIPCSocket(ctx, chainID, ipcDecisionsIdentifier, decisionEvents)
if err != nil {
return nil, err
}
return &EventSockets{
consensusSocket: consensusIPC,
decisionsSocket: decisionsIPC,
}, nil
}
// Accept delivers a message to the underlying eventSockets
func (ipcs *EventSockets) Accept(ctx *snow.Context, containerID ids.ID, container []byte) error {
if ipcs.consensusSocket != nil {
if err := ipcs.consensusSocket.Accept(ctx, containerID, container); err != nil {
return err
}
}
if ipcs.decisionsSocket != nil {
if err := ipcs.decisionsSocket.Accept(ctx, containerID, container); err != nil {
return err
}
}
return nil
}
// stop closes the underlying eventSockets
func (ipcs *EventSockets) stop() error {
errs := wrappers.Errs{}
if ipcs.consensusSocket != nil {
errs.Add(ipcs.consensusSocket.stop())
}
if ipcs.decisionsSocket != nil {
errs.Add(ipcs.decisionsSocket.stop())
}
return errs.Err
}
// ConsensusURL returns the URL of socket receiving consensus events
func (ipcs *EventSockets) ConsensusURL() string {
return ipcs.consensusSocket.URL()
}
// DecisionsURL returns the URL of socket receiving decisions events
func (ipcs *EventSockets) DecisionsURL() string {
return ipcs.decisionsSocket.URL()
}
// eventSocket is a single IPC socket for a single chain
type eventSocket struct {
url string
log logging.Logger
socket *socket.Socket
unregisterFn func() error
}
// newEventIPCSocket creates a *eventSocket for the given chain and
// EventDispatcher that writes to a local IPC socket
func newEventIPCSocket(ctx context, chainID ids.ID, name string, events *triggers.EventDispatcher) (*eventSocket, error) {
var (
url = ipcURL(ctx, chainID, name)
ipcName = ipcIdentifierPrefix + "-" + name
)
err := os.Remove(url)
if err != nil && !errors.Is(err, syscall.ENOENT) {
return nil, err
}
eis := &eventSocket{
log: ctx.log,
url: url,
socket: socket.NewSocket(url, ctx.log),
unregisterFn: func() error {
return events.DeregisterChain(chainID, ipcName)
},
}
if err := eis.socket.Listen(); err != nil {
if err := eis.socket.Close(); err != nil {
return nil, err
}
return nil, err
}
if err := events.RegisterChain(chainID, ipcName, eis, false); err != nil {
if err := eis.stop(); err != nil {
return nil, err
}
return nil, err
}
return eis, nil
}
// Accept delivers a message to the eventSocket
func (eis *eventSocket) Accept(_ *snow.Context, _ ids.ID, container []byte) error {
eis.socket.Send(container)
return nil
}
// stop unregisters the event handler and closes the eventSocket
func (eis *eventSocket) stop() error {
eis.log.Info("closing Chain IPC")
errs := wrappers.Errs{}
errs.Add(eis.unregisterFn(), eis.socket.Close())
return errs.Err
}
// URL returns the URL of the socket
func (eis *eventSocket) URL() string {
return eis.url
}