Use slice instead of map

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

.travis.yml

@@ -1,30 +0,0 @@
-os:
-  - linux
-
-language: go
-
-go:
-  - 1.12.x
-
-env:
-  global:
-    - GOTFLAGS="-race"
-  matrix:
-    - BUILD_DEPTYPE=gomod
-
-
-# disable travis install
-install:
-  - true
-
-script:
-  - bash <(curl -s https://raw.githubusercontent.com/ipfs/ci-helpers/master/travis-ci/run-standard-tests.sh)
-
-
-cache:
-  directories:
-    - $GOPATH/pkg/mod
-    - $HOME/.cache/go-build
-
-notifications:
-  email: false

README.md

@@ -0,0 +1,25 @@
+# go-eventbus
+
+[![](https://img.shields.io/badge/made%20by-Protocol%20Labs-blue.svg?style=flat-square)](https://protocol.ai)
+[![](https://img.shields.io/badge/project-libp2p-yellow.svg?style=flat-square)](https://libp2p.io/)
+[![](https://img.shields.io/badge/freenode-%23libp2p-yellow.svg?style=flat-square)](http://webchat.freenode.net/?channels=%23libp2p)
+[![GoDoc](https://godoc.org/github.com/libp2p/go-eventbus?status.svg)](https://godoc.org/github.com/libp2p/go-eventbus)
+[![Coverage Status](https://coveralls.io/repos/github/libp2p/go-eventbus/badge.svg?branch=master)](https://coveralls.io/github/libp2p/go-eventbus?branch=master)
+[![Build Status](https://travis-ci.org/libp2p/go-eventbus.svg?branch=master)](https://travis-ci.org/libp2p/go-eventbus)
+[![Discourse posts](https://img.shields.io/discourse/https/discuss.libp2p.io/posts.svg)](https://discuss.libp2p.io)
+
+> Simple and fast Go event bus
+
+## Install
+
+```sh
+go get github.com/libp2p/go-eventbus
+```
+
+## Usage
+
+Check out the [GoDocs](https://godoc.org/github.com/libp2p/go-eventbus).
+
+## License
+
+Dual-licensed under MIT and ASLv2, by way of the [Permissive License Stack](https://protocol.ai/blog/announcing-the-permissive-license-stack/).

basic.go

@@ -0,0 +1,190 @@
+package event
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+	"sync"
+	"sync/atomic"
+)
+
+///////////////////////
+// BUS
+
+type bus struct {
+	lk    sync.Mutex
+	nodes map[string]*node
+}
+
+func NewBus() Bus {
+	return &bus{
+		nodes: map[string]*node{},
+	}
+}
+
+func (b *bus) withNode(typ reflect.Type, cb func(*node)) error {
+	path := typePath(typ)
+
+	b.lk.Lock()
+
+	n, ok := b.nodes[path]
+	if !ok {
+		n = newNode(typ)
+		b.nodes[path] = n
+	}
+
+	n.lk.Lock()
+	b.lk.Unlock()
+	defer n.lk.Unlock()
+	cb(n)
+	return nil
+}
+
+func (b *bus) tryDropNode(typ reflect.Type) {
+	path := typePath(typ)
+
+	b.lk.Lock()
+	n, ok := b.nodes[path]
+	if !ok { // already dropped
+		b.lk.Unlock()
+		return
+	}
+
+	n.lk.Lock()
+	if n.nEmitters > 0 || len(n.sinks) > 0 {
+		n.lk.Unlock()
+		b.lk.Unlock()
+		return // still in use
+	}
+	n.lk.Unlock()
+
+	delete(b.nodes, path)
+	b.lk.Unlock()
+}
+
+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()
+			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(typ.Elem())
+			}
+		}
+	})
+	return
+}
+
+func (b *bus) Emitter(evtType interface{}, _ ...EmitterOption) (e EmitFunc, c CancelFunc, err error) {
+	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
+
+		e = func(event interface{}) {
+			if closed {
+				panic("emitter is closed")
+			}
+			n.emit(event)
+		}
+
+		c = func() {
+			closed = true
+			if atomic.AddInt32(&n.nEmitters, -1) == 0 {
+				b.tryDropNode(typ)
+			}
+		}
+	})
+	return
+}
+
+///////////////////////
+// NODE
+
+type node struct {
+	// Note: make sure to NEVER lock bus.lk when this lock is held
+	lk sync.RWMutex
+
+	typ reflect.Type
+
+	// emitter ref count
+	nEmitters int32
+
+	keepLast bool
+	last     reflect.Value
+
+	sinks []reflect.Value
+}
+
+func newNode(typ reflect.Type) *node {
+	return &node{
+		typ: typ,
+	}
+}
+
+func (n *node) sub(outChan reflect.Value) {
+	n.sinks = append(n.sinks, outChan)
+}
+
+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()))
+	}
+
+	n.lk.RLock()
+	// TODO: try using reflect.Select
+	for _, ch := range n.sinks {
+		ch.Send(eval)
+	}
+	n.lk.RUnlock()
+}
+
+///////////////////////
+// UTILS
+
+func typePath(t reflect.Type) string {
+	return t.PkgPath() + "/" + t.String()
+}
+
+var _ Bus = &bus{}

basic_test.go

@@ -0,0 +1,350 @@
+package event
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+type EventA struct{}
+type EventB int
+
+func (EventA) String() string {
+	return "Oh, Hello"
+}
+
+func TestEmit(t *testing.T) {
+	bus := NewBus()
+	events := make(chan EventA)
+	cancel, err := bus.Subscribe(events)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	go func() {
+		defer cancel()
+		<-events
+	}()
+
+	emit, cancel, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancel()
+
+	emit(EventA{})
+}
+
+func TestSub(t *testing.T) {
+	bus := NewBus()
+	events := make(chan EventB)
+	cancel, err := bus.Subscribe(events)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var event EventB
+
+	var wait sync.WaitGroup
+	wait.Add(1)
+
+	go func() {
+		defer cancel()
+		event = <-events
+		wait.Done()
+	}()
+
+	emit, cancel, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancel()
+
+	emit(EventB(7))
+	wait.Wait()
+
+	if event != 7 {
+		t.Error("got wrong event")
+	}
+}
+
+func TestEmitNoSubNoBlock(t *testing.T) {
+	bus := NewBus()
+
+	emit, cancel, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancel()
+
+	emit(EventA{})
+}
+
+func TestEmitOnClosed(t *testing.T) {
+	bus := NewBus()
+
+	emit, cancel, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	cancel()
+
+	defer func() {
+		r := recover()
+		if r == nil {
+			t.Errorf("expected panic")
+		}
+		if r.(string) != "emitter is closed" {
+			t.Error("unexpected message")
+		}
+	}()
+
+	emit(EventA{})
+}
+
+func TestClosingRaces(t *testing.T) {
+	subs := 50000
+	emits := 50000
+
+	var wg sync.WaitGroup
+	var lk sync.RWMutex
+	lk.Lock()
+
+	wg.Add(subs + emits)
+
+	b := NewBus()
+
+	for i := 0; i < subs; i++ {
+		go func() {
+			lk.RLock()
+			defer lk.RUnlock()
+
+			cancel, _ := b.Subscribe(make(chan EventA))
+			time.Sleep(10 * time.Millisecond)
+			cancel()
+
+			wg.Done()
+		}()
+	}
+	for i := 0; i < emits; i++ {
+		go func() {
+			lk.RLock()
+			defer lk.RUnlock()
+
+			_, cancel, _ := b.Emitter(new(EventA))
+			time.Sleep(10 * time.Millisecond)
+			cancel()
+
+			wg.Done()
+		}()
+	}
+
+	time.Sleep(10 * time.Millisecond)
+	lk.Unlock() // start everything
+
+	wg.Wait()
+
+	if len(b.(*bus).nodes) != 0 {
+		t.Error("expected no nodes")
+	}
+}
+
+func TestSubMany(t *testing.T) {
+	bus := NewBus()
+
+	var r int32
+
+	n := 50000
+	var wait sync.WaitGroup
+	var ready sync.WaitGroup
+	wait.Add(n)
+	ready.Add(n)
+
+	for i := 0; i < n; i++ {
+		go func() {
+			events := make(chan EventB)
+			cancel, err := bus.Subscribe(events)
+			if err != nil {
+				panic(err)
+			}
+			defer cancel()
+
+			ready.Done()
+			atomic.AddInt32(&r, int32(<-events))
+			wait.Done()
+		}()
+	}
+
+	emit, cancel, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancel()
+
+	ready.Wait()
+
+	emit(EventB(7))
+	wait.Wait()
+
+	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()
+
+	var r int64
+
+	var wait sync.WaitGroup
+	var ready sync.WaitGroup
+	wait.Add(subs + emits)
+	ready.Add(subs)
+
+	for i := 0; i < subs; i++ {
+		go func() {
+			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))
+			}
+			wait.Done()
+		}()
+	}
+
+	for i := 0; i < emits; i++ {
+		go func() {
+			emit, cancel, err := bus.Emitter(new(EventB))
+			if err != nil {
+				panic(err)
+			}
+			defer cancel()
+
+			ready.Wait()
+
+			for i := 0; i < msgs; i++ {
+				emit(EventB(97))
+			}
+
+			wait.Done()
+		}()
+	}
+
+	wait.Wait()
+
+	if int(r) != 97*subs*emits*msgs {
+		t.Fatal("got wrong result")
+	}
+}
+
+func TestBothMany(t *testing.T) {
+	testMany(t, 10000, 100, 10)
+}
+
+func BenchmarkSubs(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, b.N, 100, 100)
+}
+
+func BenchmarkEmits(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, 100, b.N, 100)
+}
+
+func BenchmarkMsgs(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, 100, 100, b.N)
+}
+
+func BenchmarkOneToMany(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, b.N, 1, 100)
+}
+
+func BenchmarkManyToOne(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, 1, b.N, 100)
+}
+
+func BenchmarkMs1e2m4(b *testing.B) {
+	b.N = 1000000
+	b.ReportAllocs()
+	testMany(b, 10, 100, 10000)
+}
+
+func BenchmarkMs1e0m6(b *testing.B) {
+	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()
+	testMany(b, 1, 1000000, 1)
+}
+
+func BenchmarkMs6e0m0(b *testing.B) {
+	b.N = 1000000
+	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()
+
+}

codecov.yml

@@ -1,3 +0,0 @@
-coverage:
-  range: "50...100"
-comment: off

interface.go

@@ -0,0 +1,49 @@
+package event
+
+import (
+	"errors"
+	"reflect"
+)
+
+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 channel will cause
+	// publishers to get blocked
+	Subscribe(typedChan interface{}, opts ...SubOption) (CancelFunc, error)
+
+	// Emitter creates new emitter
+	//
+	// 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()
+	Emitter(eventType interface{}, opts ...EmitterOption) (EmitFunc, CancelFunc, error)
+}
+
+// EmitFunc emits events. If any channel subscribed to the topic is blocked,
+// calls to EmitFunc will block
+//
+// Calling this function with wrong event type will cause a panic
+type EmitFunc func(event interface{})
+
+type CancelFunc func()