generify options.

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,199 @@
+package event
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+	"sync"
+	"sync/atomic"
+)
+
+///////////////////////
+// BUS
+
+type basicBus struct {
+	lk    sync.Mutex
+	nodes map[reflect.Type]*node
+}
+
+func NewBus() *basicBus {
+	return &basicBus{
+		nodes: map[reflect.Type]*node{},
+	}
+}
+
+func (b *basicBus) withNode(typ reflect.Type, cb func(*node), async func(*node)) error {
+	b.lk.Lock()
+
+	n, ok := b.nodes[typ]
+	if !ok {
+		n = newNode(typ)
+		b.nodes[typ] = n
+	}
+
+	n.lk.Lock()
+	b.lk.Unlock()
+
+	cb(n)
+
+	go func() {
+		defer n.lk.Unlock()
+		async(n)
+	}()
+
+	return nil
+}
+
+func (b *basicBus) tryDropNode(typ reflect.Type) {
+	b.lk.Lock()
+	n, ok := b.nodes[typ]
+	if !ok { // already dropped
+		b.lk.Unlock()
+		return
+	}
+
+	n.lk.Lock()
+	if atomic.LoadInt32(&n.nEmitters) > 0 || len(n.sinks) > 0 {
+		n.lk.Unlock()
+		b.lk.Unlock()
+		return // still in use
+	}
+	n.lk.Unlock()
+
+	delete(b.nodes, typ)
+	b.lk.Unlock()
+}
+
+func (b *basicBus) 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) {
+		n.sinks = append(n.sinks, 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[len(n.sinks)-1], reflect.Value{}
+					n.sinks = n.sinks[:len(n.sinks)-1]
+					break
+				}
+			}
+			tryDrop := len(n.sinks) == 0 && atomic.LoadInt32(&n.nEmitters) == 0
+			n.lk.Unlock()
+			if tryDrop {
+				b.tryDropNode(typ.Elem())
+			}
+		}
+	}, func(n *node) {
+		if n.keepLast {
+			lastVal, ok := n.last.Load().(reflect.Value)
+			if !ok {
+				return
+			}
+
+			refCh.Send(lastVal)
+		}
+	})
+	return
+}
+
+func (b *basicBus) Emitter(evtType interface{}, opts ...EmitterOption) (e EmitFunc, err error) {
+	var settings emitterSettings
+	for _, opt := range opts {
+		opt(&settings)
+	}
+
+	typ := reflect.TypeOf(evtType)
+	if typ.Kind() != reflect.Ptr {
+		return 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
+		n.keepLast = n.keepLast || settings.makeStateful
+
+		e = func(event interface{}) {
+			if closed {
+				panic("emitter is closed")
+			}
+			if event == closeEmit {
+				closed = true
+				if atomic.AddInt32(&n.nEmitters, -1) == 0 {
+					b.tryDropNode(typ)
+				}
+				return
+			}
+			n.emit(event)
+		}
+	}, func(_ *node) {})
+	return
+}
+
+///////////////////////
+// NODE
+
+type node struct {
+	// Note: make sure to NEVER lock basicBus.lk when this lock is held
+	lk sync.RWMutex
+
+	typ reflect.Type
+
+	// emitter ref count
+	nEmitters int32
+
+	keepLast bool
+	last     atomic.Value
+
+	sinks []reflect.Value
+}
+
+func newNode(typ reflect.Type) *node {
+	return &node{
+		typ: typ,
+	}
+}
+
+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()
+	if n.keepLast {
+		n.last.Store(eval)
+	}
+
+	for _, ch := range n.sinks {
+		ch.Send(eval)
+	}
+	n.lk.RUnlock()
+}
+
+///////////////////////
+// UTILS
+
+var _ Bus = &basicBus{}

basic_test.go

@@ -0,0 +1,434 @@
+package event
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/jbenet/go-detect-race"
+)
+
+type EventA struct{}
+type EventB int
+
+func getN() int {
+	n := 50000
+	if detectrace.WithRace() {
+		n = 1000
+	}
+	return n
+}
+
+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, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer emit.Close()
+
+	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, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer emit.Close()
+
+	emit(EventB(7))
+	wait.Wait()
+
+	if event != 7 {
+		t.Error("got wrong event")
+	}
+}
+
+func TestEmitNoSubNoBlock(t *testing.T) {
+	bus := NewBus()
+
+	emit, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer emit.Close()
+
+	emit(EventA{})
+}
+
+func TestEmitOnClosed(t *testing.T) {
+	bus := NewBus()
+
+	emit, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	emit.Close()
+
+	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 := getN()
+	emits := getN()
+
+	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()
+
+			emit, _ := b.Emitter(new(EventA))
+			time.Sleep(10 * time.Millisecond)
+			emit.Close()
+
+			wg.Done()
+		}()
+	}
+
+	time.Sleep(10 * time.Millisecond)
+	lk.Unlock() // start everything
+
+	wg.Wait()
+
+	if len(b.nodes) != 0 {
+		t.Error("expected no nodes")
+	}
+}
+
+func TestSubMany(t *testing.T) {
+	bus := NewBus()
+
+	var r int32
+
+	n := getN()
+	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, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer emit.Close()
+
+	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, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer emit.Close()
+
+	emit(EventA{})
+	wait.Wait()
+
+	if event.String() != "Oh, Hello" {
+		t.Error("didn't get the correct message")
+	}
+}
+
+func TestNonStateful(t *testing.T) {
+	bus := NewBus()
+	emit, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer emit.Close()
+
+	eventsA := make(chan EventB, 1)
+	cancelS, err := bus.Subscribe(eventsA)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancelS()
+
+	select {
+	case <-eventsA:
+		t.Fatal("didn't expect to get an event")
+	default:
+	}
+
+	emit(EventB(1))
+
+	select {
+	case e := <-eventsA:
+		if e != 1 {
+			t.Fatal("got wrong event")
+		}
+	default:
+		t.Fatal("expected to get an event")
+	}
+
+	eventsB := make(chan EventB, 1)
+	cancelS2, err := bus.Subscribe(eventsB)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancelS2()
+
+	select {
+	case <-eventsA:
+		t.Fatal("didn't expect to get an event")
+	default:
+	}
+}
+
+func TestStateful(t *testing.T) {
+	bus := NewBus()
+	emit, err := bus.Emitter(new(EventB), Stateful)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer emit.Close()
+
+	emit(EventB(2))
+
+	eventsA := make(chan EventB, 1)
+	cancelS, err := bus.Subscribe(eventsA)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancelS()
+
+	if <-eventsA != 2 {
+		t.Fatal("got wrong event")
+	}
+}
+
+func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
+	if detectrace.WithRace() && subs+emits > 5000 {
+		t.SkipNow()
+	}
+
+	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, err := bus.Emitter(new(EventB), func(settings *emitterSettings) {
+				settings.makeStateful = stateful
+			})
+			if err != nil {
+				panic(err)
+			}
+			defer emit.Close()
+
+			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, false)
+}
+
+func BenchmarkSubs(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, b.N, 100, 100, false)
+}
+
+func BenchmarkEmits(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, 100, b.N, 100, false)
+}
+
+func BenchmarkMsgs(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, 100, 100, b.N, false)
+}
+
+func BenchmarkOneToMany(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, b.N, 1, 100, false)
+}
+
+func BenchmarkManyToOne(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, 1, b.N, 100, false)
+}
+
+func BenchmarkMs1e2m4(b *testing.B) {
+	b.N = 1000000
+	b.ReportAllocs()
+	testMany(b, 10, 100, 10000, false)
+}
+
+func BenchmarkMs1e0m6(b *testing.B) {
+	b.N = 10000000
+	b.ReportAllocs()
+	testMany(b, 10, 1, 1000000, false)
+}
+
+func BenchmarkMs0e0m6(b *testing.B) {
+	b.N = 1000000
+	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)
+}

codecov.yml

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

go.mod

@@ -1,3 +1,5 @@
 module github.com/libp2p/go-eventbus
 
 go 1.12
+
+require github.com/jbenet/go-detect-race v0.0.0-20150302022421-3463798d9574

go.sum

@@ -0,0 +1,2 @@
+github.com/jbenet/go-detect-race v0.0.0-20150302022421-3463798d9574 h1:Pxjl8Wn3cCU7nB/MCmPEUMbjMHxXFqODW6rce0jpxB4=
+github.com/jbenet/go-detect-race v0.0.0-20150302022421-3463798d9574/go.mod h1:gynVu6LUw+xMXD3XEvjHQcIbJkWEamnGjJDebRHqTd0=

interface.go

@@ -0,0 +1,96 @@
+package event
+
+import (
+	"errors"
+	"reflect"
+)
+
+var closeEmit struct{}
+
+type subSettings struct {
+	forcedType reflect.Type
+}
+
+type SubOption func(interface{}) error
+
+// ForceSubType is a Subscribe option which overrides the type to which
+// the subscription will be done. Note that the evtType must be assignable
+// to channel type.
+//
+// This also allows for subscribing to multiple eventbus channels with one
+// Go channel to get better ordering guarantees.
+//
+// Example:
+// type Event struct{}
+// func (Event) String() string {
+//    return "event"
+// }
+//
+// eventCh := make(chan fmt.Stringer) // interface { String() string }
+// cancel, err := eventbus.Subscribe(eventCh, event.ForceSubType(new(Event)))
+// [...]
+func ForceSubType(evtType interface{}) SubOption {
+	return func(settings interface{}) error {
+		s := settings.(*subSettings)
+		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 {
+	makeStateful bool
+}
+type EmitterOption func(interface{}) error
+
+// Stateful is an Emitter option which makes makes the eventbus channel
+// 'remember' last event sent, and when a new subscriber joins the
+// bus, the remembered event is immediately sent to the subscription
+// channel.
+//
+// This allows to provide state tracking for dynamic systems, and/or
+// allows new subscribers to verify that there are Emitters on the channel
+func Stateful(s *emitterSettings) {
+	s.makeStateful = true
+}
+
+// Bus is an interface to type-based event delivery system
+type Bus interface {
+	// Subscribe creates new subscription. Failing to drain the channel will cause
+	// publishers to get blocked. CancelFunc is guaranteed to return after last send
+	// to the channel
+	//
+	// Example:
+	// ch := make(chan EventT, 10)
+	// defer close(ch)
+	// cancel, err := eventbus.Subscribe(ch)
+	// defer cancel()
+	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:
+	// emit, err := eventbus.Emitter(new(EventT))
+	// defer emit.Close() // MUST call this after being done with the emitter
+	//
+	// emit(EventT{})
+	Emitter(eventType interface{}, opts ...EmitterOption) (EmitFunc, 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{})
+
+func (f EmitFunc) Close() {
+	f(closeEmit)
+}
+
+type CancelFunc func()