POC SendTo

basic.go

@@ -12,7 +12,7 @@ import (
 // BUS
 
 type bus struct {
-	lk    sync.Mutex
+	lk sync.Mutex
 	nodes map[string]*node
 }
 
@@ -56,7 +56,7 @@ func (b *bus) tryDropNode(evtType interface{}) {
 	}
 
 	n.lk.Lock()
-	if n.nEmitters > 0 || n.sinkLen() > 0 {
+	if n.nEmitters > 0 || len(n.sinks) > 0 {
 		n.lk.Unlock()
 		b.lk.Unlock()
 		return // still in use
@@ -69,13 +69,15 @@ func (b *bus) tryDropNode(evtType interface{}) {
 
 func (b *bus) Subscribe(evtType interface{}, _ ...SubOption) (s <-chan interface{}, c CancelFunc, err error) {
 	err = b.withNode(evtType, func(n *node) {
+		// when all subs are waiting on this channel, setting this to 1 doesn't
+		// really affect benchmarks
 		out, i := n.sub(0)
 		s = out
 		c = func() {
 			n.lk.Lock()
-			n.sink.Delete(i)
+			delete(n.sinks, i)
 			close(out)
-			tryDrop := n.sinkLen() == 0 && n.nEmitters == 0
+			tryDrop := len(n.sinks) == 0 && n.nEmitters == 0
 			n.lk.Unlock()
 			if tryDrop {
 				b.tryDropNode(evtType)
@@ -107,13 +109,34 @@ func (b *bus) Emitter(evtType interface{}, _ ...EmitterOption) (e EmitFunc, c Ca
 	return
 }
 
+func (b *bus) SendTo(typedChan interface{}) (CancelFunc, error) {
+	typ := reflect.TypeOf(typedChan)
+	if typ.Kind() != reflect.Chan {
+		return nil, errors.New("expected a channel")
+	}
+	if typ.ChanDir() & reflect.SendDir == 0 {
+		return nil, errors.New("channel doesn't allow send")
+	}
+	etype := reflect.New(typ.Elem())
+	sub, cf, err := b.Subscribe(etype.Interface())
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		tcv := reflect.ValueOf(typedChan)
+		for event := range sub {
+			tcv.Send(reflect.ValueOf(event))
+		}
+	}()
+
+	return cf, nil
+}
+
 ///////////////////////
 // NODE
 
 type node struct {
-	// not under lock
-	sink sync.Map
-
 	// Note: make sure to NEVER lock bus.lk when this lock is held
 	lk sync.RWMutex
 
@@ -123,29 +146,26 @@ type node struct {
 	nEmitters int32
 
 	// sink index counter
-	sinkC int
+	sinkC     int
+
+	// TODO: we could make emit a bit faster by making this into an array, but
+	//  it doesn't seem needed for now
+	sinks  map[int]chan interface{}
 }
 
 func newNode(typ reflect.Type) *node {
 	return &node{
 		typ: typ,
-	}
-}
 
-func (n *node) sinkLen() int {
-	ln := 0
-	n.sink.Range(func(_, _ interface{}) bool {
-		ln = ln + 1
-		return true
-	})
-	return ln
+		sinks: map[int]chan interface{}{},
+	}
 }
 
 func (n *node) sub(buf int) (chan interface{}, int) {
 	out := make(chan interface{}, buf)
 	i := n.sinkC
 	n.sinkC++
-	n.sink.Store(i, out)
+	n.sinks[i] = out
 	return out, i
 }
 
@@ -155,10 +175,11 @@ func (n *node) emit(event interface{}) {
 		panic(fmt.Sprintf("Emit called with wrong type. expected: %s, got: %s", n.typ, etype))
 	}
 
-	n.sink.Range(func(_, ch interface{}) bool {
-		ch.(chan interface{}) <- event
-		return true
-	})
+	n.lk.RLock()
+	for _, ch := range n.sinks {
+		ch <- event
+	}
+	n.lk.RUnlock()
 }
 
 ///////////////////////

basic_test.go

@@ -185,6 +185,42 @@ func TestSubMany(t *testing.T) {
 	}
 }
 
+func TestSendTo(t *testing.T) {
+	testSendTo(t, 1000)
+}
+
+func testSendTo(t testing.TB, msgs int) {
+	bus := NewBus()
+
+	go func() {
+		emit, cancel, err := bus.Emitter(new(EventB))
+		if err != nil {
+			panic(err)
+		}
+		defer cancel()
+
+		for i := 0; i < msgs; i++ {
+			emit(EventB(97))
+		}
+	}()
+
+	ch := make(chan EventB)
+	cancel, err := bus.SendTo(ch)
+	if err != nil {
+		return
+	}
+	defer cancel()
+
+	r := 0
+	for i := 0; i < msgs; i++ {
+		r += int(<-ch)
+	}
+
+	if int(r) != 97 * msgs {
+		t.Fatal("got wrong result")
+	}
+}
+
 func testMany(t testing.TB, subs, emits, msgs int) {
 	bus := NewBus()
 
@@ -272,11 +308,17 @@ func BenchmarkMs1e2m4(b *testing.B) {
 }
 
 func BenchmarkMs1e0m6(b *testing.B) {
-	b.N = 1000000
+	b.N = 10000000
 	b.ReportAllocs()
 	testMany(b, 10, 1, 1000000)
 }
 
+func BenchmarkMs0e0m6(b *testing.B) {
+	b.N = 1000000
+	b.ReportAllocs()
+	testMany(b, 1, 1, 1000000)
+}
+
 func BenchmarkMs0e6m0(b *testing.B) {
 	b.N = 1000000
 	b.ReportAllocs()
@@ -288,3 +330,9 @@ func BenchmarkMs6e0m0(b *testing.B) {
 	b.ReportAllocs()
 	testMany(b, 1000000, 1, 1)
 }
+
+func BenchmarkSendTo(b *testing.B) {
+	b.N = 1000000
+	b.ReportAllocs()
+	testSendTo(b, b.N)
+}

interface.go

@@ -20,6 +20,7 @@ type Bus interface {
 	// evt := (<-sub).(os.Signal) // guaranteed to be safe
 	Subscribe(eventType interface{}, opts ...SubOption) (<-chan interface{}, CancelFunc, error)
 
+	SendTo(typedChan interface{}) (CancelFunc, error)
 
 	Emitter(eventType interface{}, opts ...EmitterOption) (EmitFunc, CancelFunc, error)
 }