generated from RTradeLtd/repo-template
-
Notifications
You must be signed in to change notification settings - Fork 0
/
basic.go
263 lines (215 loc) · 4.82 KB
/
basic.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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
package eventbus
import (
"errors"
"fmt"
"reflect"
"sync"
"sync/atomic"
"github.com/RTradeLtd/libp2px-core/event"
)
///////////////////////
// BUS
// basicBus is a type-based event delivery system
type basicBus struct {
lk sync.Mutex
nodes map[reflect.Type]*node
}
var _ event.Bus = (*basicBus)(nil)
type emitter struct {
n *node
typ reflect.Type
closed int32
dropper func(reflect.Type)
}
func (e *emitter) Emit(evt interface{}) error {
if atomic.LoadInt32(&e.closed) != 0 {
return fmt.Errorf("emitter is closed")
}
e.n.emit(evt)
return nil
}
func (e *emitter) Close() error {
if !atomic.CompareAndSwapInt32(&e.closed, 0, 1) {
return fmt.Errorf("closed an emitter more than once")
}
if atomic.AddInt32(&e.n.nEmitters, -1) == 0 {
e.dropper(e.typ)
}
return nil
}
func NewBus() event.Bus {
return &basicBus{
nodes: map[reflect.Type]*node{},
}
}
func (b *basicBus) withNode(typ reflect.Type, cb func(*node), async func(*node)) {
b.lk.Lock()
n, ok := b.nodes[typ]
if !ok {
n = newNode(typ)
b.nodes[typ] = n
}
n.lk.Lock()
b.lk.Unlock()
cb(n)
if async == nil {
n.lk.Unlock()
} else {
go func() {
defer n.lk.Unlock()
async(n)
}()
}
}
func (b *basicBus) tryDropNode(typ reflect.Type) {
b.lk.Lock()
n, ok := b.nodes[typ]
if !ok { // already dropped
b.lk.Unlock()
return
}
n.lk.Lock()
if atomic.LoadInt32(&n.nEmitters) > 0 || len(n.sinks) > 0 {
n.lk.Unlock()
b.lk.Unlock()
return // still in use
}
n.lk.Unlock()
delete(b.nodes, typ)
b.lk.Unlock()
}
type sub struct {
ch chan interface{}
nodes []*node
dropper func(reflect.Type)
}
func (s *sub) Out() <-chan interface{} {
return s.ch
}
func (s *sub) Close() error {
go func() {
// drain the event channel, will return when closed and drained.
// this is necessary to unblock publishes to this channel.
for range s.ch {
}
}()
for _, n := range s.nodes {
n.lk.Lock()
for i := 0; i < len(n.sinks); i++ {
if n.sinks[i] == s.ch {
n.sinks[i], n.sinks[len(n.sinks)-1] = n.sinks[len(n.sinks)-1], nil
n.sinks = n.sinks[:len(n.sinks)-1]
break
}
}
tryDrop := len(n.sinks) == 0 && atomic.LoadInt32(&n.nEmitters) == 0
n.lk.Unlock()
if tryDrop {
s.dropper(n.typ)
}
}
close(s.ch)
return nil
}
var _ event.Subscription = (*sub)(nil)
// Subscribe creates new subscription. Failing to drain the channel will cause
// publishers to get blocked. CancelFunc is guaranteed to return after last send
// to the channel
func (b *basicBus) Subscribe(evtTypes interface{}, opts ...event.SubscriptionOpt) (_ event.Subscription, err error) {
settings := subSettings(subSettingsDefault)
for _, opt := range opts {
if err := opt(&settings); err != nil {
return nil, err
}
}
types, ok := evtTypes.([]interface{})
if !ok {
types = []interface{}{evtTypes}
}
out := &sub{
ch: make(chan interface{}, settings.buffer),
nodes: make([]*node, len(types)),
dropper: b.tryDropNode,
}
for _, etyp := range types {
if reflect.TypeOf(etyp).Kind() != reflect.Ptr {
return nil, errors.New("subscribe called with non-pointer type")
}
}
for i, etyp := range types {
typ := reflect.TypeOf(etyp)
b.withNode(typ.Elem(), func(n *node) {
n.sinks = append(n.sinks, out.ch)
out.nodes[i] = n
}, func(n *node) {
if n.keepLast {
l := n.last
if l == nil {
return
}
out.ch <- l
}
})
}
return out, nil
}
// Emitter creates new emitter
//
// eventType accepts typed nil pointers, and uses the type information to
// select output type
//
// Example:
// emit, err := eventbus.Emitter(new(EventT))
// defer emit.Close() // MUST call this after being done with the emitter
//
// emit(EventT{})
func (b *basicBus) Emitter(evtType interface{}, opts ...event.EmitterOpt) (e event.Emitter, err error) {
var settings emitterSettings
for _, opt := range opts {
if err := opt(&settings); err != nil {
return nil, err
}
}
typ := reflect.TypeOf(evtType)
if typ.Kind() != reflect.Ptr {
return nil, errors.New("emitter called with non-pointer type")
}
typ = typ.Elem()
b.withNode(typ, func(n *node) {
atomic.AddInt32(&n.nEmitters, 1)
n.keepLast = n.keepLast || settings.makeStateful
e = &emitter{n: n, typ: typ, dropper: b.tryDropNode}
}, nil)
return
}
///////////////////////
// NODE
type node struct {
// Note: make sure to NEVER lock basicBus.lk when this lock is held
lk sync.Mutex
typ reflect.Type
// emitter ref count
nEmitters int32
keepLast bool
last interface{}
sinks []chan interface{}
}
func newNode(typ reflect.Type) *node {
return &node{
typ: typ,
}
}
func (n *node) emit(event interface{}) {
typ := reflect.TypeOf(event)
if typ != n.typ {
panic(fmt.Sprintf("Emit called with wrong type. expected: %s, got: %s", n.typ, typ))
}
n.lk.Lock()
if n.keepLast {
n.last = event
}
for _, ch := range n.sinks {
ch <- event
}
n.lk.Unlock()
}