Use slice instead of map

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

basic.go

@@ -12,7 +12,7 @@ import (
 // BUS
 
 type bus struct {
-	lk sync.Mutex
+	lk    sync.Mutex
 	nodes map[string]*node
 }
 
@@ -22,12 +22,7 @@ func NewBus() Bus {
 	}
 }
 
-func (b *bus) withNode(evtType interface{}, cb func(*node)) error {
-	typ := reflect.TypeOf(evtType)
-	if typ.Kind() != reflect.Ptr {
-		return errors.New("subscribe called with non-pointer type")
-	}
-	typ = typ.Elem()
+func (b *bus) withNode(typ reflect.Type, cb func(*node)) error {
 	path := typePath(typ)
 
 	b.lk.Lock()
@@ -45,8 +40,8 @@ func (b *bus) withNode(evtType interface{}, cb func(*node)) error {
 	return nil
 }
 
-func (b *bus) tryDropNode(evtType interface{}) {
-	path := typePath(reflect.TypeOf(evtType).Elem())
+func (b *bus) tryDropNode(typ reflect.Type) {
+	path := typePath(typ)
 
 	b.lk.Lock()
 	n, ok := b.nodes[path]
@@ -67,18 +62,48 @@ func (b *bus) tryDropNode(evtType interface{}) {
 	b.lk.Unlock()
 }
 
-func (b *bus) Subscribe(evtType interface{}, _ ...SubOption) (s <-chan interface{}, c CancelFunc, err error) {
-	err = b.withNode(evtType, func(n *node) {
-		out, i := n.sub(0)
-		s = out
+func (b *bus) Subscribe(typedChan interface{}, opts ...SubOption) (c CancelFunc, err error) {
+	var settings SubSettings
+	for _, opt := range opts {
+		if err := opt(&settings); err != nil {
+			return nil, err
+		}
+	}
+
+	refCh := reflect.ValueOf(typedChan)
+	typ := refCh.Type()
+	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")
+	}
+
+	if settings.forcedType != nil {
+		if settings.forcedType.Elem().AssignableTo(typ) {
+			return nil, fmt.Errorf("forced type %s cannot be sent to chan %s", settings.forcedType, typ)
+		}
+		typ = settings.forcedType
+	}
+
+	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
+		n.sub(refCh)
+
 		c = func() {
 			n.lk.Lock()
-			delete(n.sinks, i)
-			close(out)
+			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 {
-				b.tryDropNode(evtType)
+				b.tryDropNode(typ.Elem())
 			}
 		}
 	})
@@ -86,7 +111,13 @@ func (b *bus) Subscribe(evtType interface{}, _ ...SubOption) (s <-chan interface
 }
 
 func (b *bus) Emitter(evtType interface{}, _ ...EmitterOption) (e EmitFunc, c CancelFunc, err error) {
-	err = b.withNode(evtType, func(n *node) {
+	typ := reflect.TypeOf(evtType)
+	if typ.Kind() != reflect.Ptr {
+		return nil, nil, errors.New("emitter called with non-pointer type")
+	}
+	typ = typ.Elem()
+
+	err = b.withNode(typ, func(n *node) {
 		atomic.AddInt32(&n.nEmitters, 1)
 		closed := false
 
@@ -100,7 +131,7 @@ func (b *bus) Emitter(evtType interface{}, _ ...EmitterOption) (e EmitFunc, c Ca
 		c = func() {
 			closed = true
 			if atomic.AddInt32(&n.nEmitters, -1) == 0 {
-				b.tryDropNode(evtType)
+				b.tryDropNode(typ)
 			}
 		}
 	})
@@ -119,39 +150,32 @@ type node struct {
 	// emitter ref count
 	nEmitters int32
 
-	// sink index counter
-	sinkC     int
+	keepLast bool
+	last     reflect.Value
 
-	// 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{}
+	sinks []reflect.Value
 }
 
 func newNode(typ reflect.Type) *node {
 	return &node{
 		typ: typ,
-
-		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.sinks[i] = out
-	return out, i
+func (n *node) sub(outChan reflect.Value) {
+	n.sinks = append(n.sinks, outChan)
 }
 
 func (n *node) emit(event interface{}) {
-	etype := reflect.TypeOf(event)
-	if etype != n.typ {
-		panic(fmt.Sprintf("Emit called with wrong type. expected: %s, got: %s", n.typ, etype))
+	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()))
 	}
 
 	n.lk.RLock()
+	// TODO: try using reflect.Select
 	for _, ch := range n.sinks {
-		ch <- event
+		ch.Send(eval)
 	}
 	n.lk.RUnlock()
 }

basic_test.go

@@ -1,6 +1,7 @@
 package event
 
 import (
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -10,9 +11,14 @@ import (
 type EventA struct{}
 type EventB int
 
+func (EventA) String() string {
+	return "Oh, Hello"
+}
+
 func TestEmit(t *testing.T) {
 	bus := NewBus()
-	events, cancel, err := bus.Subscribe(new(EventA))
+	events := make(chan EventA)
+	cancel, err := bus.Subscribe(events)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -33,7 +39,8 @@ func TestEmit(t *testing.T) {
 
 func TestSub(t *testing.T) {
 	bus := NewBus()
-	events, cancel, err := bus.Subscribe(new(EventB))
+	events := make(chan EventB)
+	cancel, err := bus.Subscribe(events)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -45,7 +52,7 @@ func TestSub(t *testing.T) {
 
 	go func() {
 		defer cancel()
-		event = (<-events).(EventB)
+		event = <-events
 		wait.Done()
 	}()
 
@@ -114,7 +121,7 @@ func TestClosingRaces(t *testing.T) {
 			lk.RLock()
 			defer lk.RUnlock()
 
-			_, cancel, _ := b.Subscribe(new(EventA))
+			cancel, _ := b.Subscribe(make(chan EventA))
 			time.Sleep(10 * time.Millisecond)
 			cancel()
 
@@ -157,14 +164,15 @@ func TestSubMany(t *testing.T) {
 
 	for i := 0; i < n; i++ {
 		go func() {
-			events, cancel, err := bus.Subscribe(new(EventB))
+			events := make(chan EventB)
+			cancel, err := bus.Subscribe(events)
 			if err != nil {
 				panic(err)
 			}
 			defer cancel()
 
 			ready.Done()
-			atomic.AddInt32(&r, int32((<-events).(EventB)))
+			atomic.AddInt32(&r, int32(<-events))
 			wait.Done()
 		}()
 	}
@@ -180,11 +188,44 @@ 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")
 	}
 }
 
+func TestSubType(t *testing.T) {
+	bus := NewBus()
+	events := make(chan fmt.Stringer)
+	cancel, err := bus.Subscribe(events, ForceSubType(new(EventA)))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var event fmt.Stringer
+
+	var wait sync.WaitGroup
+	wait.Add(1)
+
+	go func() {
+		defer cancel()
+		event = <-events
+		wait.Done()
+	}()
+
+	emit, cancel, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancel()
+
+	emit(EventA{})
+	wait.Wait()
+
+	if event.String() != "Oh, Hello" {
+		t.Error("didn't get the correct message")
+	}
+}
+
 func testMany(t testing.TB, subs, emits, msgs int) {
 	bus := NewBus()
 
@@ -197,15 +238,16 @@ func testMany(t testing.TB, subs, emits, msgs int) {
 
 	for i := 0; i < subs; i++ {
 		go func() {
-			events, cancel, err := bus.Subscribe(new(EventB))
+			events := make(chan EventB)
+			cancel, err := bus.Subscribe(events)
 			if err != nil {
 				panic(err)
 			}
 			defer cancel()
 
 			ready.Done()
-			for i := 0; i < emits * msgs; i++ {
-				atomic.AddInt64(&r, int64((<-events).(EventB)))
+			for i := 0; i < emits*msgs; i++ {
+				atomic.AddInt64(&r, int64(<-events))
 			}
 			wait.Done()
 		}()
@@ -231,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")
 	}
 }
@@ -272,11 +314,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 +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

@@ -1,26 +1,42 @@
 package event
 
-type SubSettings struct {}
-type SubOption func(*SubSettings)
+import (
+	"errors"
+	"reflect"
+)
 
-type EmitterSettings struct {}
+type SubSettings struct {
+	forcedType reflect.Type
+}
+type SubOption func(*SubSettings) error
+
+func ForceSubType(evtType interface{}) SubOption {
+	return func(s *SubSettings) error {
+		typ := reflect.TypeOf(evtType)
+		if typ.Kind() != reflect.Ptr {
+			return errors.New("ForceSubType called with non-pointer type")
+		}
+		s.forcedType = typ
+		return nil
+	}
+}
+
+type EmitterSettings struct{}
 type EmitterOption func(*EmitterSettings)
 
 type Bus interface {
-	// Subscribe creates new subscription. Failing to drain the incoming channel
-	// will cause publishers to get blocked
+	// Subscribe creates new subscription. Failing to drain the channel will cause
+	// 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(eventType interface{}, opts ...SubOption) (<-chan interface{}, CancelFunc, error)
-
-
 	Emitter(eventType interface{}, opts ...EmitterOption) (EmitFunc, CancelFunc, error)
 }
 
@@ -31,4 +47,3 @@ type Bus interface {
 type EmitFunc func(event interface{})
 
 type CancelFunc func()
-