POC SendTo

basic.go

@@ -69,6 +69,8 @@ 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() {
@@ -107,6 +109,30 @@ 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
 

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)
 }