Use slice instead of map

License: MIT
Signed-off-by: Jakub Sztandera <kubuxu@protonmail.ch>

basic.go

@@ -75,7 +75,7 @@ func (b *bus) Subscribe(typedChan interface{}, opts ...SubOption) (c CancelFunc,
 	if typ.Kind() != reflect.Chan {
 		return nil, errors.New("expected a channel")
 	}
-	if typ.ChanDir() & reflect.SendDir == 0 {
+	if typ.ChanDir()&reflect.SendDir == 0 {
 		return nil, errors.New("channel doesn't allow send")
 	}
 
@@ -89,10 +89,17 @@ func (b *bus) Subscribe(typedChan interface{}, opts ...SubOption) (c CancelFunc,
 	err = b.withNode(typ.Elem(), func(n *node) {
 		// when all subs are waiting on this channel, setting this to 1 doesn't
 		// really affect benchmarks
-		i := n.sub(refCh)
+		n.sub(refCh)
+
 		c = func() {
 			n.lk.Lock()
-			delete(n.sinks, i)
+			for i := 0; i < len(n.sinks); i++ {
+				if n.sinks[i] == refCh {
+					n.sinks[i] = n.sinks[len(n.sinks)-1]
+					n.sinks = n.sinks[:len(n.sinks)-1]
+					break
+				}
+			}
 			tryDrop := len(n.sinks) == 0 && n.nEmitters == 0
 			n.lk.Unlock()
 			if tryDrop {
@@ -143,27 +150,20 @@ type node struct {
 	// emitter ref count
 	nEmitters int32
 
-	// sink index counter
+	keepLast bool
-	sinkC     int
+	last     reflect.Value
 
-	// TODO: we could make emit a bit faster by making this into an array, but
+	sinks []reflect.Value
-	//  it doesn't seem needed for now
-	sinks  map[int]reflect.Value
 }
 
 func newNode(typ reflect.Type) *node {
 	return &node{
 		typ: typ,
-
-		sinks: map[int]reflect.Value{},
 	}
 }
 
-func (n *node) sub(outChan reflect.Value) int {
+func (n *node) sub(outChan reflect.Value) {
-	i := n.sinkC
+	n.sinks = append(n.sinks, outChan)
-	n.sinkC++
-	n.sinks[i] = outChan
-	return i
 }
 
 func (n *node) emit(event interface{}) {

basic_test.go

@@ -188,7 +188,7 @@ func TestSubMany(t *testing.T) {
 	emit(EventB(7))
 	wait.Wait()
 
-	if int(r) != 7 * n {
+	if int(r) != 7*n {
 		t.Error("got wrong result")
 	}
 }
@@ -246,7 +246,7 @@ func testMany(t testing.TB, subs, emits, msgs int) {
 			defer cancel()
 
 			ready.Done()
-			for i := 0; i < emits * msgs; i++ {
+			for i := 0; i < emits*msgs; i++ {
 				atomic.AddInt64(&r, int64(<-events))
 			}
 			wait.Done()
@@ -273,7 +273,7 @@ func testMany(t testing.TB, subs, emits, msgs int) {
 
 	wait.Wait()
 
-	if int(r) != 97 * subs * emits * msgs {
+	if int(r) != 97*subs*emits*msgs {
 		t.Fatal("got wrong result")
 	}
 }
@@ -336,3 +336,15 @@ func BenchmarkMs6e0m0(b *testing.B) {
 	b.ReportAllocs()
 	testMany(b, 1000000, 1, 1)
 }
+
+func t() {
+	bus := NewBus()
+
+	events := make(chan fmt.Stringer)
+	cancel, err := bus.Subscribe(events, Stateful)
+	if err != nil {
+		//
+	}
+	defer cancel()
+
+}

interface.go

@@ -21,23 +21,22 @@ func ForceSubType(evtType interface{}) SubOption {
 	}
 }
 
-type EmitterSettings struct {}
+type EmitterSettings struct{}
 type EmitterOption func(*EmitterSettings)
 
 type Bus interface {
-	// Subscribe creates new subscription. Failing to drain the incoming channel
+	// Subscribe creates new subscription. Failing to drain the channel will cause
-	// will cause publishers to get blocked
+	// publishers to get blocked
+	Subscribe(typedChan interface{}, opts ...SubOption) (CancelFunc, error)
+
+	// Emitter creates new emitter
 	//
-	// evtTypes only accepts typed nil pointers, and uses the type information to
+	// eventType accepts typed nil pointers, and uses the type information to
 	// select output type
 	//
 	// Example:
 	// sub, cancel, err := eventbus.Subscribe(new(os.Signal))
 	// defer cancel()
-	//
-	// evt := (<-sub).(os.Signal) // guaranteed to be safe
-	Subscribe(typedChan interface{}, opts ...SubOption) (CancelFunc, error)
-
 	Emitter(eventType interface{}, opts ...EmitterOption) (EmitFunc, CancelFunc, error)
 }