mirror of
				https://source.quilibrium.com/quilibrium/ceremonyclient.git
				synced 2025-11-04 00:57:26 +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)
 | 
						|
	}
 | 
						|
}
 |