mirror of
https://source.quilibrium.com/quilibrium/ceremonyclient.git
synced 2024-11-14 04:05:17 +00:00
419 lines
8.5 KiB
Go
419 lines
8.5 KiB
Go
|
package eventbus
|
||
|
|
||
|
import (
|
||
|
"errors"
|
||
|
"fmt"
|
||
|
"reflect"
|
||
|
"sync"
|
||
|
"sync/atomic"
|
||
|
|
||
|
"github.com/libp2p/go-libp2p/core/event"
|
||
|
)
|
||
|
|
||
|
// /////////////////////
|
||
|
// BUS
|
||
|
|
||
|
// basicBus is a type-based event delivery system
|
||
|
type basicBus struct {
|
||
|
lk sync.RWMutex
|
||
|
nodes map[reflect.Type]*node
|
||
|
wildcard *wildcardNode
|
||
|
metricsTracer MetricsTracer
|
||
|
}
|
||
|
|
||
|
var _ event.Bus = (*basicBus)(nil)
|
||
|
|
||
|
type emitter struct {
|
||
|
n *node
|
||
|
w *wildcardNode
|
||
|
typ reflect.Type
|
||
|
closed atomic.Bool
|
||
|
dropper func(reflect.Type)
|
||
|
metricsTracer MetricsTracer
|
||
|
}
|
||
|
|
||
|
func (e *emitter) Emit(evt interface{}) error {
|
||
|
if e.closed.Load() {
|
||
|
return fmt.Errorf("emitter is closed")
|
||
|
}
|
||
|
|
||
|
e.n.emit(evt)
|
||
|
e.w.emit(evt)
|
||
|
|
||
|
if e.metricsTracer != nil {
|
||
|
e.metricsTracer.EventEmitted(e.typ)
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (e *emitter) Close() error {
|
||
|
if !e.closed.CompareAndSwap(false, true) {
|
||
|
return fmt.Errorf("closed an emitter more than once")
|
||
|
}
|
||
|
if e.n.nEmitters.Add(-1) == 0 {
|
||
|
e.dropper(e.typ)
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func NewBus(opts ...Option) event.Bus {
|
||
|
bus := &basicBus{
|
||
|
nodes: map[reflect.Type]*node{},
|
||
|
wildcard: &wildcardNode{},
|
||
|
}
|
||
|
for _, opt := range opts {
|
||
|
opt(bus)
|
||
|
}
|
||
|
return bus
|
||
|
}
|
||
|
|
||
|
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.metricsTracer)
|
||
|
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 n.nEmitters.Load() > 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 wildcardSub struct {
|
||
|
ch chan interface{}
|
||
|
w *wildcardNode
|
||
|
metricsTracer MetricsTracer
|
||
|
name string
|
||
|
}
|
||
|
|
||
|
func (w *wildcardSub) Out() <-chan interface{} {
|
||
|
return w.ch
|
||
|
}
|
||
|
|
||
|
func (w *wildcardSub) Close() error {
|
||
|
w.w.removeSink(w.ch)
|
||
|
if w.metricsTracer != nil {
|
||
|
w.metricsTracer.RemoveSubscriber(reflect.TypeOf(event.WildcardSubscription))
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (w *wildcardSub) Name() string {
|
||
|
return w.name
|
||
|
}
|
||
|
|
||
|
type namedSink struct {
|
||
|
name string
|
||
|
ch chan interface{}
|
||
|
}
|
||
|
|
||
|
type sub struct {
|
||
|
ch chan interface{}
|
||
|
nodes []*node
|
||
|
dropper func(reflect.Type)
|
||
|
metricsTracer MetricsTracer
|
||
|
name string
|
||
|
}
|
||
|
|
||
|
func (s *sub) Name() string {
|
||
|
return s.name
|
||
|
}
|
||
|
|
||
|
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].ch == 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]
|
||
|
|
||
|
if s.metricsTracer != nil {
|
||
|
s.metricsTracer.RemoveSubscriber(n.typ)
|
||
|
}
|
||
|
break
|
||
|
}
|
||
|
}
|
||
|
|
||
|
tryDrop := len(n.sinks) == 0 && n.nEmitters.Load() == 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 := newSubSettings()
|
||
|
for _, opt := range opts {
|
||
|
if err := opt(&settings); err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
}
|
||
|
|
||
|
if evtTypes == event.WildcardSubscription {
|
||
|
out := &wildcardSub{
|
||
|
ch: make(chan interface{}, settings.buffer),
|
||
|
w: b.wildcard,
|
||
|
metricsTracer: b.metricsTracer,
|
||
|
name: settings.name,
|
||
|
}
|
||
|
b.wildcard.addSink(&namedSink{ch: out.ch, name: out.name})
|
||
|
return out, nil
|
||
|
}
|
||
|
|
||
|
types, ok := evtTypes.([]interface{})
|
||
|
if !ok {
|
||
|
types = []interface{}{evtTypes}
|
||
|
}
|
||
|
|
||
|
if len(types) > 1 {
|
||
|
for _, t := range types {
|
||
|
if t == event.WildcardSubscription {
|
||
|
return nil, fmt.Errorf("wildcard subscriptions must be started separately")
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
|
||
|
out := &sub{
|
||
|
ch: make(chan interface{}, settings.buffer),
|
||
|
nodes: make([]*node, len(types)),
|
||
|
|
||
|
dropper: b.tryDropNode,
|
||
|
metricsTracer: b.metricsTracer,
|
||
|
name: settings.name,
|
||
|
}
|
||
|
|
||
|
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, &namedSink{ch: out.ch, name: out.name})
|
||
|
out.nodes[i] = n
|
||
|
if b.metricsTracer != nil {
|
||
|
b.metricsTracer.AddSubscriber(typ.Elem())
|
||
|
}
|
||
|
}, 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) {
|
||
|
if evtType == event.WildcardSubscription {
|
||
|
return nil, fmt.Errorf("illegal emitter for wildcard subscription")
|
||
|
}
|
||
|
|
||
|
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) {
|
||
|
n.nEmitters.Add(1)
|
||
|
n.keepLast = n.keepLast || settings.makeStateful
|
||
|
e = &emitter{n: n, typ: typ, dropper: b.tryDropNode, w: b.wildcard, metricsTracer: b.metricsTracer}
|
||
|
}, nil)
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// GetAllEventTypes returns all the event types that this bus has emitters
|
||
|
// or subscribers for.
|
||
|
func (b *basicBus) GetAllEventTypes() []reflect.Type {
|
||
|
b.lk.RLock()
|
||
|
defer b.lk.RUnlock()
|
||
|
|
||
|
types := make([]reflect.Type, 0, len(b.nodes))
|
||
|
for t := range b.nodes {
|
||
|
types = append(types, t)
|
||
|
}
|
||
|
return types
|
||
|
}
|
||
|
|
||
|
// /////////////////////
|
||
|
// NODE
|
||
|
|
||
|
type wildcardNode struct {
|
||
|
sync.RWMutex
|
||
|
nSinks atomic.Int32
|
||
|
sinks []*namedSink
|
||
|
metricsTracer MetricsTracer
|
||
|
}
|
||
|
|
||
|
func (n *wildcardNode) addSink(sink *namedSink) {
|
||
|
n.nSinks.Add(1) // ok to do outside the lock
|
||
|
n.Lock()
|
||
|
n.sinks = append(n.sinks, sink)
|
||
|
n.Unlock()
|
||
|
|
||
|
if n.metricsTracer != nil {
|
||
|
n.metricsTracer.AddSubscriber(reflect.TypeOf(event.WildcardSubscription))
|
||
|
}
|
||
|
}
|
||
|
|
||
|
func (n *wildcardNode) removeSink(ch chan interface{}) {
|
||
|
n.nSinks.Add(-1) // ok to do outside the lock
|
||
|
n.Lock()
|
||
|
for i := 0; i < len(n.sinks); i++ {
|
||
|
if n.sinks[i].ch == 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
|
||
|
}
|
||
|
}
|
||
|
n.Unlock()
|
||
|
}
|
||
|
|
||
|
func (n *wildcardNode) emit(evt interface{}) {
|
||
|
if n.nSinks.Load() == 0 {
|
||
|
return
|
||
|
}
|
||
|
|
||
|
n.RLock()
|
||
|
for _, sink := range n.sinks {
|
||
|
|
||
|
// Sending metrics before sending on channel allows us to
|
||
|
// record channel full events before blocking
|
||
|
sendSubscriberMetrics(n.metricsTracer, sink)
|
||
|
|
||
|
sink.ch <- evt
|
||
|
}
|
||
|
n.RUnlock()
|
||
|
}
|
||
|
|
||
|
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 atomic.Int32
|
||
|
|
||
|
keepLast bool
|
||
|
last interface{}
|
||
|
|
||
|
sinks []*namedSink
|
||
|
metricsTracer MetricsTracer
|
||
|
}
|
||
|
|
||
|
func newNode(typ reflect.Type, metricsTracer MetricsTracer) *node {
|
||
|
return &node{
|
||
|
typ: typ,
|
||
|
metricsTracer: metricsTracer,
|
||
|
}
|
||
|
}
|
||
|
|
||
|
func (n *node) emit(evt interface{}) {
|
||
|
typ := reflect.TypeOf(evt)
|
||
|
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 = evt
|
||
|
}
|
||
|
|
||
|
for _, sink := range n.sinks {
|
||
|
|
||
|
// Sending metrics before sending on channel allows us to
|
||
|
// record channel full events before blocking
|
||
|
sendSubscriberMetrics(n.metricsTracer, sink)
|
||
|
sink.ch <- evt
|
||
|
}
|
||
|
n.lk.Unlock()
|
||
|
}
|
||
|
|
||
|
func sendSubscriberMetrics(metricsTracer MetricsTracer, sink *namedSink) {
|
||
|
if metricsTracer != nil {
|
||
|
metricsTracer.SubscriberQueueLength(sink.name, len(sink.ch)+1)
|
||
|
metricsTracer.SubscriberQueueFull(sink.name, len(sink.ch)+1 >= cap(sink.ch))
|
||
|
metricsTracer.SubscriberEventQueued(sink.name)
|
||
|
}
|
||
|
}
|