fix: serialize publishing

Ensure that all subscribers see events in the same order. This also ensures that
the subscribers never see the initial "latest" event after some other event.

fixes #16

basic.go

@@ -104,9 +104,16 @@ func (s *sub) Out() <-chan interface{} {
 }
 
 func (s *sub) Close() error {
-	close(s.ch)
+	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
@@ -114,12 +121,16 @@ func (s *sub) Close() error {
 				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
 }
 
@@ -129,7 +140,7 @@ var _ event.Subscription = (*sub)(nil)
 // 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) {
-	var settings subSettings
+	settings := subSettings(subSettingsDefault)
 	for _, opt := range opts {
 		if err := opt(&settings); err != nil {
 			return nil, err
@@ -148,19 +159,21 @@ func (b *basicBus) Subscribe(evtTypes interface{}, opts ...event.SubscriptionOpt
 		dropper: b.tryDropNode,
 	}
 
-	for i, etyp := range types {
-		typ := reflect.TypeOf(etyp)
-
-		if typ.Kind() != reflect.Ptr {
+	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)
 
 		err = 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.Load()
+				l := n.last
 				if l == nil {
 					return
 				}
@@ -184,6 +197,7 @@ func (b *basicBus) Subscribe(evtTypes interface{}, opts ...event.SubscriptionOpt
 // 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
@@ -209,7 +223,7 @@ func (b *basicBus) Emitter(evtType interface{}, opts ...event.EmitterOpt) (e eve
 
 type node struct {
 	// Note: make sure to NEVER lock basicBus.lk when this lock is held
-	lk sync.RWMutex
+	lk sync.Mutex
 
 	typ reflect.Type
 
@@ -217,7 +231,7 @@ type node struct {
 	nEmitters int32
 
 	keepLast bool
-	last     atomic.Value
+	last     interface{}
 
 	sinks []chan interface{}
 }
@@ -229,18 +243,18 @@ func newNode(typ reflect.Type) *node {
 }
 
 func (n *node) emit(event interface{}) {
-	eval := reflect.ValueOf(event)
-	if eval.Type() != n.typ {
-		panic(fmt.Sprintf("Emit called with wrong type. expected: %s, got: %s", n.typ, eval.Type()))
+	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.RLock()
+	n.lk.Lock()
 	if n.keepLast {
-		n.last.Store(event)
+		n.last = event
 	}
 
 	for _, ch := range n.sinks {
 		ch <- event
 	}
-	n.lk.RUnlock()
+	n.lk.Unlock()
 }

basic_test.go

@@ -25,6 +25,17 @@ func (EventA) String() string {
 	return "Oh, Hello"
 }
 
+func TestDefaultSubIsBuffered(t *testing.T) {
+	bus := NewBus()
+	s, err := bus.Subscribe(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if cap(s.(*sub).ch) == 0 {
+		t.Fatalf("without any options subscribe should be buffered. was %d", cap(s.(*sub).ch))
+	}
+}
+
 func TestEmit(t *testing.T) {
 	bus := NewBus()
 	sub, err := bus.Subscribe(new(EventA))
@@ -297,6 +308,49 @@ func TestStateful(t *testing.T) {
 	}
 }
 
+func TestCloseBlocking(t *testing.T) {
+	bus := NewBus()
+	em, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sub, err := bus.Subscribe(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	go func() {
+		em.Emit(EventB(159))
+	}()
+
+	time.Sleep(10 * time.Millisecond) // make sure that emit is blocked
+
+	sub.Close() // cancel sub
+}
+
+func panicOnTimeout(d time.Duration) {
+	<-time.After(d)
+	panic("timeout reached")
+}
+
+func TestSubFailFully(t *testing.T) {
+	bus := NewBus()
+	em, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = bus.Subscribe([]interface{}{new(EventB), 5})
+	if err == nil || err.Error() != "subscribe called with non-pointer type" {
+		t.Fatal(err)
+	}
+
+	go panicOnTimeout(5 * time.Second)
+
+	em.Emit(EventB(159)) // will hang if sub doesn't fail properly
+}
+
 func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
 	if race.WithRace() && subs+emits > 5000 {
 		t.SkipNow()
@@ -363,69 +417,122 @@ func TestBothMany(t *testing.T) {
 	testMany(t, 10000, 100, 10, false)
 }
 
-func BenchmarkSubs(b *testing.B) {
-	b.ReportAllocs()
-	testMany(b, b.N, 100, 100, false)
+type benchCase struct {
+	subs     int
+	emits    int
+	stateful bool
 }
 
-func BenchmarkEmits(b *testing.B) {
-	b.ReportAllocs()
-	testMany(b, 100, b.N, 100, false)
+func (bc benchCase) name() string {
+	return fmt.Sprintf("subs-%03d/emits-%03d/stateful-%t", bc.subs, bc.emits, bc.stateful)
 }
 
-func BenchmarkMsgs(b *testing.B) {
-	b.ReportAllocs()
-	testMany(b, 100, 100, b.N, false)
+func genTestCases() []benchCase {
+	ret := make([]benchCase, 0, 200)
+	for stateful := 0; stateful < 2; stateful++ {
+		for subs := uint(0); subs <= 8; subs = subs + 4 {
+			for emits := uint(0); emits <= 8; emits = emits + 4 {
+				ret = append(ret, benchCase{1 << subs, 1 << emits, stateful == 1})
+			}
+		}
+	}
+	return ret
 }
 
-func BenchmarkOneToMany(b *testing.B) {
-	b.ReportAllocs()
-	testMany(b, b.N, 1, 100, false)
+func BenchmarkEvents(b *testing.B) {
+	for _, bc := range genTestCases() {
+		b.Run(bc.name(), benchMany(bc))
+	}
 }
 
-func BenchmarkManyToOne(b *testing.B) {
-	b.ReportAllocs()
-	testMany(b, 1, b.N, 100, false)
+func benchMany(bc benchCase) func(*testing.B) {
+	return func(b *testing.B) {
+		b.ReportAllocs()
+		subs := bc.subs
+		emits := bc.emits
+		stateful := bc.stateful
+		bus := NewBus()
+		var wait sync.WaitGroup
+		var ready sync.WaitGroup
+		wait.Add(subs + emits)
+		ready.Add(subs + emits)
+
+		for i := 0; i < subs; i++ {
+			go func() {
+				sub, err := bus.Subscribe(new(EventB))
+				if err != nil {
+					panic(err)
+				}
+				defer sub.Close()
+
+				ready.Done()
+				ready.Wait()
+				for i := 0; i < (b.N/emits)*emits; i++ {
+					_, ok := <-sub.Out()
+					if !ok {
+						panic("wat")
+					}
+				}
+				wait.Done()
+			}()
+		}
+
+		for i := 0; i < emits; i++ {
+			go func() {
+				em, err := bus.Emitter(new(EventB), func(settings interface{}) error {
+					settings.(*emitterSettings).makeStateful = stateful
+					return nil
+				})
+				if err != nil {
+					panic(err)
+				}
+				defer em.Close()
+
+				ready.Done()
+				ready.Wait()
+
+				for i := 0; i < b.N/emits; i++ {
+					em.Emit(EventB(97))
+				}
+
+				wait.Done()
+			}()
+		}
+		ready.Wait()
+		b.ResetTimer()
+		wait.Wait()
+	}
 }
 
-func BenchmarkMs1e2m4(b *testing.B) {
-	b.N = 1000000
+var div = 100
+
+func BenchmarkSubscribe(b *testing.B) {
 	b.ReportAllocs()
-	testMany(b, 10, 100, 10000, false)
+	for i := 0; i < b.N/div; i++ {
+		bus := NewBus()
+		for j := 0; j < div; j++ {
+			bus.Subscribe(new(EventA))
+		}
+	}
 }
 
-func BenchmarkMs1e0m6(b *testing.B) {
-	b.N = 10000000
+func BenchmarkEmitter(b *testing.B) {
 	b.ReportAllocs()
-	testMany(b, 10, 1, 1000000, false)
+	for i := 0; i < b.N/div; i++ {
+		bus := NewBus()
+		for j := 0; j < div; j++ {
+			bus.Emitter(new(EventA))
+		}
+	}
 }
 
-func BenchmarkMs0e0m6(b *testing.B) {
-	b.N = 1000000
+func BenchmarkSubscribeAndEmitter(b *testing.B) {
 	b.ReportAllocs()
-	testMany(b, 1, 1, 1000000, false)
-}
-
-func BenchmarkStatefulMs1e0m6(b *testing.B) {
-	b.N = 10000000
-	b.ReportAllocs()
-	testMany(b, 10, 1, 1000000, true)
-}
-
-func BenchmarkStatefulMs0e0m6(b *testing.B) {
-	b.N = 1000000
-	b.ReportAllocs()
-	testMany(b, 1, 1, 1000000, true)
-}
-
-func BenchmarkMs0e6m0(b *testing.B) {
-	b.N = 1000000
-	b.ReportAllocs()
-	testMany(b, 1, 1000000, 1, false)
-}
-
-func BenchmarkMs6e0m0(b *testing.B) {
-	b.N = 1000000
-	b.ReportAllocs()
-	testMany(b, 1000000, 1, 1, false)
+	for i := 0; i < b.N/div; i++ {
+		bus := NewBus()
+		for j := 0; j < div; j++ {
+			bus.Subscribe(new(EventA))
+			bus.Emitter(new(EventA))
+		}
+	}
 }

opts.go

@@ -4,6 +4,10 @@ type subSettings struct {
 	buffer int
 }
 
+var subSettingsDefault = subSettings{
+	buffer: 16,
+}
+
 func BufSize(n int) func(interface{}) error {
 	return func(s interface{}) error {
 		s.(*subSettings).buffer = n