reference basis

.travis.yml

+os:
+  - linux
+
+language: go
+
+go:
+  - 1.13.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

LICENSE-APACHE

+Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

LICENSE-MIT

+The MIT License (MIT)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

README.md

 # go-eventbus
 
-dms3 p2p go-eventbus
\ No newline at end of file
+[![](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.com/libp2p/go-eventbus.svg?branch=master)](https://travis-ci.com/libp2p/go-eventbus)
+[![Discourse posts](https://img.shields.io/discourse/https/discuss.libp2p.io/posts.svg)](https://discuss.libp2p.io)
+
+> Simple and fast eventbus for type-based local event delivery.
+
+## 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

+package eventbus
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+	"sync"
+	"sync/atomic"
+
+	"github.com/libp2p/go-libp2p-core/event"
+)
+
+///////////////////////
+// BUS
+
+// basicBus is a type-based event delivery system
+type basicBus struct {
+	lk       sync.RWMutex
+	nodes    map[reflect.Type]*node
+	wildcard *wildcardNode
+}
+
+var _ event.Bus = (*basicBus)(nil)
+
+type emitter struct {
+	n       *node
+	w       *wildcardNode
+	typ     reflect.Type
+	closed  int32
+	dropper func(reflect.Type)
+}
+
+func (e *emitter) Emit(evt interface{}) error {
+	if atomic.LoadInt32(&e.closed) != 0 {
+		return fmt.Errorf("emitter is closed")
+	}
+	e.n.emit(evt)
+	e.w.emit(evt)
+
+	return nil
+}
+
+func (e *emitter) Close() error {
+	if !atomic.CompareAndSwapInt32(&e.closed, 0, 1) {
+		return fmt.Errorf("closed an emitter more than once")
+	}
+	if atomic.AddInt32(&e.n.nEmitters, -1) == 0 {
+		e.dropper(e.typ)
+	}
+	return nil
+}
+
+func NewBus() event.Bus {
+	return &basicBus{
+		nodes:    map[reflect.Type]*node{},
+		wildcard: new(wildcardNode),
+	}
+}
+
+func (b *basicBus) withNode(typ reflect.Type, cb func(*node), async func(*node)) {
+	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)
+
+	if async == nil {
+		n.lk.Unlock()
+	} else {
+		go func() {
+			defer n.lk.Unlock()
+			async(n)
+		}()
+	}
+}
+
+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()
+}
+
+type wildcardSub struct {
+	ch chan interface{}
+	w  *wildcardNode
+}
+
+func (w *wildcardSub) Out() <-chan interface{} {
+	return w.ch
+}
+
+func (w *wildcardSub) Close() error {
+	w.w.removeSink(w.ch)
+	return nil
+}
+
+type sub struct {
+	ch      chan interface{}
+	nodes   []*node
+	dropper func(reflect.Type)
+}
+
+func (s *sub) Out() <-chan interface{} {
+	return s.ch
+}
+
+func (s *sub) Close() error {
+	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
+				n.sinks = n.sinks[:len(n.sinks)-1]
+				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
+}
+
+var _ event.Subscription = (*sub)(nil)
+
+// 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
+func (b *basicBus) Subscribe(evtTypes interface{}, opts ...event.SubscriptionOpt) (_ event.Subscription, err error) {
+	settings := subSettingsDefault
+	for _, opt := range opts {
+		if err := opt(&settings); err != nil {
+			return nil, err
+		}
+	}
+
+	if evtTypes == event.WildcardSubscription {
+		out := &wildcardSub{
+			ch: make(chan interface{}, settings.buffer),
+			w:  b.wildcard,
+		}
+		b.wildcard.addSink(out.ch)
+		return out, nil
+	}
+
+	types, ok := evtTypes.([]interface{})
+	if !ok {
+		types = []interface{}{evtTypes}
+	}
+
+	if len(types) > 1 {
+		for _, t := range types {
+			if t == event.WildcardSubscription {
+				return nil, fmt.Errorf("wildcard subscriptions must be started separately")
+			}
+		}
+	}
+
+	out := &sub{
+		ch:    make(chan interface{}, settings.buffer),
+		nodes: make([]*node, len(types)),
+
+		dropper: b.tryDropNode,
+	}
+
+	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)
+
+		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
+				if l == nil {
+					return
+				}
+				out.ch <- l
+			}
+		})
+	}
+
+	return out, nil
+}
+
+// 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{})
+func (b *basicBus) Emitter(evtType interface{}, opts ...event.EmitterOpt) (e event.Emitter, err error) {
+	if evtType == event.WildcardSubscription {
+		return nil, fmt.Errorf("illegal emitter for wildcard subscription")
+	}
+
+	var settings emitterSettings
+	for _, opt := range opts {
+		if err := opt(&settings); err != nil {
+			return nil, err
+		}
+	}
+
+	typ := reflect.TypeOf(evtType)
+	if typ.Kind() != reflect.Ptr {
+		return nil, errors.New("emitter called with non-pointer type")
+	}
+	typ = typ.Elem()
+
+	b.withNode(typ, func(n *node) {
+		atomic.AddInt32(&n.nEmitters, 1)
+		n.keepLast = n.keepLast || settings.makeStateful
+		e = &emitter{n: n, typ: typ, dropper: b.tryDropNode, w: b.wildcard}
+	}, nil)
+	return
+}
+
+// GetAllEventTypes returns all the event types that this bus has emitters
+// or subscribers for.
+func (b *basicBus) GetAllEventTypes() []reflect.Type {
+	b.lk.RLock()
+	defer b.lk.RUnlock()
+
+	types := make([]reflect.Type, 0, len(b.nodes))
+	for t, _ := range b.nodes {
+		types = append(types, t)
+	}
+	return types
+}
+
+///////////////////////
+// NODE
+
+type wildcardNode struct {
+	sync.RWMutex
+	nSinks int32
+	sinks  []chan interface{}
+}
+
+func (n *wildcardNode) addSink(ch chan interface{}) {
+	atomic.AddInt32(&n.nSinks, 1) // ok to do outside the lock
+	n.Lock()
+	n.sinks = append(n.sinks, ch)
+	n.Unlock()
+}
+
+func (n *wildcardNode) removeSink(ch chan interface{}) {
+	atomic.AddInt32(&n.nSinks, -1) // ok to do outside the lock
+	n.Lock()
+	for i := 0; i < len(n.sinks); i++ {
+		if n.sinks[i] == ch {
+			n.sinks[i], n.sinks[len(n.sinks)-1] = n.sinks[len(n.sinks)-1], nil
+			n.sinks = n.sinks[:len(n.sinks)-1]
+			break
+		}
+	}
+	n.Unlock()
+}
+
+func (n *wildcardNode) emit(evt interface{}) {
+	if atomic.LoadInt32(&n.nSinks) == 0 {
+		return
+	}
+
+	n.RLock()
+	for _, ch := range n.sinks {
+		ch <- evt
+	}
+	n.RUnlock()
+}
+
+type node struct {
+	// Note: make sure to NEVER lock basicBus.lk when this lock is held
+	lk sync.Mutex
+
+	typ reflect.Type
+
+	// emitter ref count
+	nEmitters int32
+
+	keepLast bool
+	last     interface{}
+
+	sinks []chan interface{}
+}
+
+func newNode(typ reflect.Type) *node {
+	return &node{
+		typ: typ,
+	}
+}
+
+func (n *node) emit(evt interface{}) {
+	typ := reflect.TypeOf(evt)
+	if typ != n.typ {
+		panic(fmt.Sprintf("Emit called with wrong type. expected: %s, got: %s", n.typ, typ))
+	}
+
+	n.lk.Lock()
+	if n.keepLast {
+		n.last = evt
+	}
+
+	for _, ch := range n.sinks {
+		ch <- evt
+	}
+	n.lk.Unlock()
+}

basic_test.go

+package eventbus
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/event"
+	"github.com/libp2p/go-libp2p-testing/race"
+
+	"github.com/stretchr/testify/require"
+)
+
+type EventA struct{}
+type EventB int
+
+func getN() int {
+	n := 50000
+	if race.WithRace() {
+		n = 1000
+	}
+	return n
+}
+
+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))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	go func() {
+		defer sub.Close()
+		<-sub.Out()
+	}()
+
+	em, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer em.Close()
+
+	em.Emit(EventA{})
+}
+
+func TestSub(t *testing.T) {
+	bus := NewBus()
+	sub, err := bus.Subscribe(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var event EventB
+
+	var wait sync.WaitGroup
+	wait.Add(1)
+
+	go func() {
+		defer sub.Close()
+		event = (<-sub.Out()).(EventB)
+		wait.Done()
+	}()
+
+	em, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer em.Close()
+
+	em.Emit(EventB(7))
+	wait.Wait()
+
+	if event != 7 {
+		t.Error("got wrong event")
+	}
+}
+
+func TestGetAllEventTypes(t *testing.T) {
+	bus := NewBus()
+	require.Empty(t, bus.GetAllEventTypes())
+
+	// the wildcard subscription should be returned.
+	_, err := bus.Subscribe(event.WildcardSubscription)
+	require.NoError(t, err)
+
+	_, err = bus.Subscribe(new(EventB))
+	require.NoError(t, err)
+
+	evts := bus.GetAllEventTypes()
+	require.Len(t, evts, 1)
+	require.Equal(t, reflect.TypeOf((*EventB)(nil)).Elem(), evts[0])
+
+	_, err = bus.Emitter(new(EventA))
+	require.NoError(t, err)
+
+	evts = bus.GetAllEventTypes()
+	require.Len(t, evts, 2)
+	require.Contains(t, evts, reflect.TypeOf((*EventB)(nil)).Elem())
+	require.Contains(t, evts, reflect.TypeOf((*EventA)(nil)).Elem())
+}
+
+func TestEmitNoSubNoBlock(t *testing.T) {
+	bus := NewBus()
+
+	em, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer em.Close()
+
+	em.Emit(EventA{})
+}
+
+func TestEmitOnClosed(t *testing.T) {
+	bus := NewBus()
+
+	em, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	em.Close()
+	err = em.Emit(EventA{})
+	if err == nil {
+		t.Errorf("expected error")
+	}
+	if err.Error() != "emitter is closed" {
+		t.Error("unexpected message")
+	}
+}
+
+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()
+
+			sub, _ := b.Subscribe(new(EventA))
+			time.Sleep(10 * time.Millisecond)
+			sub.Close()
+
+			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.(*basicBus).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() {
+			sub, err := bus.Subscribe(new(EventB))
+			if err != nil {
+				panic(err)
+			}
+			defer sub.Close()
+
+			ready.Done()
+			atomic.AddInt32(&r, int32((<-sub.Out()).(EventB)))
+			wait.Done()
+		}()
+	}
+
+	em, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer em.Close()
+
+	ready.Wait()
+
+	em.Emit(EventB(7))
+	wait.Wait()
+
+	if int(r) != 7*n {
+		t.Error("got wrong result")
+	}
+}
+
+func TestWildcardSubscription(t *testing.T) {
+	bus := NewBus()
+	sub, err := bus.Subscribe(event.WildcardSubscription)
+	require.NoError(t, err)
+	defer sub.Close()
+
+	em1, err := bus.Emitter(new(EventA))
+	require.NoError(t, err)
+	defer em1.Close()
+
+	em2, err := bus.Emitter(new(EventB))
+	require.NoError(t, err)
+	defer em2.Close()
+
+	require.NoError(t, em1.Emit(EventA{}))
+	require.NoError(t, em2.Emit(EventB(1)))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	var evts []interface{}
+
+LOOP:
+	for {
+		select {
+		case evt := <-sub.Out():
+			if evta, ok := evt.(EventA); ok {
+				evts = append(evts, evta)
+			}
+
+			if evtb, ok := evt.(EventB); ok {
+				evts = append(evts, evtb)
+			}
+
+			if len(evts) == 2 {
+				break LOOP
+			}
+
+		case <-ctx.Done():
+			t.Fatalf("did not receive events")
+		}
+	}
+}
+
+func TestManyWildcardSubscriptions(t *testing.T) {
+	bus := NewBus()
+	var subs []event.Subscription
+	for i := 0; i < 10; i++ {
+		sub, err := bus.Subscribe(event.WildcardSubscription)
+		require.NoError(t, err)
+		subs = append(subs, sub)
+	}
+
+	em1, err := bus.Emitter(new(EventA))
+	require.NoError(t, err)
+	defer em1.Close()
+
+	em2, err := bus.Emitter(new(EventB))
+	require.NoError(t, err)
+	defer em2.Close()
+
+	require.NoError(t, em1.Emit(EventA{}))
+	require.NoError(t, em2.Emit(EventB(1)))
+
+	// all 10 subscriptions received all 2 events.
+	for _, s := range subs {
+		require.Len(t, s.Out(), 2)
+	}
+
+	// close the first five subscriptions.
+	for _, s := range subs[:5] {
+		require.NoError(t, s.Close())
+	}
+
+	// emit another 2 events.
+	require.NoError(t, em1.Emit(EventA{}))
+	require.NoError(t, em2.Emit(EventB(1)))
+
+	// the first five still have 2 events, while the other five have 4 events.
+	for _, s := range subs[:5] {
+		require.Len(t, s.Out(), 2)
+	}
+
+	for _, s := range subs[5:] {
+		require.Len(t, s.Out(), 4)
+	}
+
+	// close them all, the first five will be closed twice (asserts idempotency).
+	for _, s := range subs {
+		require.NoError(t, s.Close())
+	}
+}
+
+func TestWildcardValidations(t *testing.T) {
+	bus := NewBus()
+
+	_, err := bus.Subscribe([]interface{}{event.WildcardSubscription, new(EventA), new(EventB)})
+	require.Error(t, err)
+
+	_, err = bus.Emitter(event.WildcardSubscription)
+	require.Error(t, err)
+}
+
+func TestSubType(t *testing.T) {
+	bus := NewBus()
+	sub, err := bus.Subscribe([]interface{}{new(EventA), new(EventB)})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var event fmt.Stringer
+
+	var wait sync.WaitGroup
+	wait.Add(1)
+
+	go func() {
+		defer sub.Close()
+		event = (<-sub.Out()).(EventA)
+		wait.Done()
+	}()
+
+	em, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer em.Close()
+
+	em.Emit(EventA{})
+	wait.Wait()
+
+	if event.String() != "Oh, Hello" {
+		t.Error("didn't get the correct message")
+	}
+}
+
+func TestNonStateful(t *testing.T) {
+	bus := NewBus()
+	em, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer em.Close()
+
+	sub1, err := bus.Subscribe(new(EventB), BufSize(1))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sub1.Close()
+
+	select {
+	case <-sub1.Out():
+		t.Fatal("didn't expect to get an event")
+	default:
+	}
+
+	em.Emit(EventB(1))
+
+	select {
+	case e := <-sub1.Out():
+		if e.(EventB) != 1 {
+			t.Fatal("got wrong event")
+		}
+	default:
+		t.Fatal("expected to get an event")
+	}
+
+	sub2, err := bus.Subscribe(new(EventB), BufSize(1))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sub2.Close()
+
+	select {
+	case <-sub2.Out():
+		t.Fatal("didn't expect to get an event")
+	default:
+	}
+}
+
+func TestStateful(t *testing.T) {
+	bus := NewBus()
+	em, err := bus.Emitter(new(EventB), Stateful)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer em.Close()
+
+	em.Emit(EventB(2))
+
+	sub, err := bus.Subscribe(new(EventB), BufSize(1))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sub.Close()
+
+	if (<-sub.Out()).(EventB) != 2 {
+		t.Fatal("got wrong event")
+	}
+}
+
+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()
+	}
+
+	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() {
+			sub, err := bus.Subscribe(new(EventB))
+			if err != nil {
+				panic(err)
+			}
+			defer sub.Close()
+
+			ready.Done()
+			for i := 0; i < emits*msgs; i++ {
+				e, ok := <-sub.Out()
+				if !ok {
+					panic("wat")
+				}
+				atomic.AddInt64(&r, int64(e.(EventB)))
+			}
+			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.Wait()
+
+			for i := 0; i < msgs; i++ {
+				em.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)
+}
+
+type benchCase struct {
+	subs     int
+	emits    int
+	stateful bool
+}
+
+func (bc benchCase) name() string {
+	return fmt.Sprintf("subs-%03d/emits-%03d/stateful-%t", bc.subs, bc.emits, bc.stateful)
+}
+
+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 BenchmarkEvents(b *testing.B) {
+	for _, bc := range genTestCases() {
+		b.Run(bc.name(), benchMany(bc))
+	}
+}
+
+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()
+	}
+}
+
+var div = 100
+
+func BenchmarkSubscribe(b *testing.B) {
+	b.ReportAllocs()
+	for i := 0; i < b.N/div; i++ {
+		bus := NewBus()
+		for j := 0; j < div; j++ {
+			bus.Subscribe(new(EventA))
+		}
+	}
+}
+
+func BenchmarkEmitter(b *testing.B) {
+	b.ReportAllocs()
+	for i := 0; i < b.N/div; i++ {
+		bus := NewBus()
+		for j := 0; j < div; j++ {
+			bus.Emitter(new(EventA))
+		}
+	}
+}
+
+func BenchmarkSubscribeAndEmitter(b *testing.B) {
+	b.ReportAllocs()
+	for i := 0; i < b.N/div; i++ {
+		bus := NewBus()
+		for j := 0; j < div; j++ {
+			bus.Subscribe(new(EventA))
+			bus.Emitter(new(EventA))
+		}
+	}
+}

codecov.yml

+coverage:
+  range: "50...100"
+comment: off

go.mod

+module github.com/libp2p/go-eventbus
+
+go 1.12
+
+require (
+	github.com/libp2p/go-libp2p-core v0.5.7
+	github.com/libp2p/go-libp2p-testing v0.1.1
+	github.com/stretchr/testify v1.4.0
+)

go.sum

+cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
+github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
+github.com/btcsuite/btcd v0.0.0-20190523000118-16327141da8c/go.mod h1:3J08xEfcugPacsc34/LKRU2yO7YmuT8yt28J8k2+rrI=
+github.com/btcsuite/btcd v0.20.1-beta h1:Ik4hyJqN8Jfyv3S4AGBOmyouMsYE3EdYODkMbQjwPGw=
+github.com/btcsuite/btcd v0.20.1-beta/go.mod h1:wVuoA8VJLEcwgqHBwHmzLRazpKxTv13Px/pDuV7OomQ=
+github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f/go.mod h1:TdznJufoqS23FtqVCzL0ZqgP5MqXbb4fg/WgDys70nA=
+github.com/btcsuite/btcutil v0.0.0-20190425235716-9e5f4b9a998d/go.mod h1:+5NJ2+qvTyV9exUAL/rxXi3DcLg2Ts+ymUAY5y4NvMg=
+github.com/btcsuite/go-socks v0.0.0-20170105172521-4720035b7bfd/go.mod h1:HHNXQzUsZCxOoE+CPiyCTO6x34Zs86zZUiwtpXoGdtg=
+github.com/btcsuite/goleveldb v0.0.0-20160330041536-7834afc9e8cd/go.mod h1:F+uVaaLLH7j4eDXPRvw78tMflu7Ie2bzYOH4Y8rRKBY=
+github.com/btcsuite/snappy-go v0.0.0-20151229074030-0bdef8d06723/go.mod h1:8woku9dyThutzjeg+3xrA5iCpBRH8XEEg3lh6TiUghc=
+github.com/btcsuite/websocket v0.0.0-20150119174127-31079b680792/go.mod h1:ghJtEyQwv5/p4Mg4C0fgbePVuGr935/5ddU9Z3TmDRY=
+github.com/btcsuite/winsvc v1.0.0/go.mod h1:jsenWakMcC0zFBFurPLEAyrnc/teJEM1O46fmI40EZs=
+github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
+github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
+github.com/davecgh/go-spew v0.0.0-20171005155431-ecdeabc65495/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
+github.com/gogo/protobuf v1.3.1 h1:DqDEcV5aeaTmdFBePNpYsp3FlcVH/2ISVVM9Qf8PSls=
+github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o=
+github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
+github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
+github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
+github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
+github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/gxed/hashland/keccakpg v0.0.1/go.mod h1:kRzw3HkwxFU1mpmPP8v1WyQzwdGfmKFJ6tItnhQ67kU=
+github.com/gxed/hashland/murmur3 v0.0.1/go.mod h1:KjXop02n4/ckmZSnY2+HKcLud/tcmvhST0bie/0lS48=
+github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
+github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
+github.com/ipfs/go-cid v0.0.2/go.mod h1:GHWU/WuQdMPmIosc4Yn1bcCT7dSeX4lBafM7iqUPQvM=
+github.com/ipfs/go-cid v0.0.5 h1:o0Ix8e/ql7Zb5UVUJEUfjsWCIY8t48++9lR8qi6oiJU=
+github.com/ipfs/go-cid v0.0.5/go.mod h1:plgt+Y5MnOey4vO4UlUazGqdbEXuFYitED67FexhXog=
+github.com/jbenet/go-cienv v0.1.0/go.mod h1:TqNnHUmJgXau0nCzC7kXWeotg3J9W34CUv5Djy1+FlA=
+github.com/jbenet/goprocess v0.1.3/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZlqdZVfqY4=
+github.com/jbenet/goprocess v0.1.4 h1:DRGOFReOMqqDNXwW70QkacFW0YN9QnwLV0Vqk+3oU0o=
+github.com/jbenet/goprocess v0.1.4/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZlqdZVfqY4=
+github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
+github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
+github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlTRt3OuAQ=
+github.com/kami-zh/go-capturer v0.0.0-20171211120116-e492ea43421d/go.mod h1:P2viExyCEfeWGU259JnaQ34Inuec4R38JCyBx2edgD0=
+github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
+github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
+github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4=
+github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/libp2p/go-buffer-pool v0.0.1/go.mod h1:xtyIz9PMobb13WaxR6Zo1Pd1zXJKYg0a8KiIvDp3TzQ=
+github.com/libp2p/go-buffer-pool v0.0.2 h1:QNK2iAFa8gjAe1SPz6mHSMuCcjs+X1wlHzeOSqcmlfs=
+github.com/libp2p/go-buffer-pool v0.0.2/go.mod h1:MvaB6xw5vOrDl8rYZGLFdKAuk/hRoRZd1Vi32+RXyFM=
+github.com/libp2p/go-flow-metrics v0.0.1/go.mod h1:Iv1GH0sG8DtYN3SVJ2eG221wMiNpZxBdp967ls1g+k8=
+github.com/libp2p/go-flow-metrics v0.0.3/go.mod h1:HeoSNUrOJVK1jEpDqVEiUOIXqhbnS27omG0uWU5slZs=
+github.com/libp2p/go-libp2p-core v0.2.0/go.mod h1:X0eyB0Gy93v0DZtSYbEM7RnMChm9Uv3j7yRXjO77xSI=
+github.com/libp2p/go-libp2p-core v0.5.7 h1:QK3xRwFxqd0Xd9bSZL+8yZ8ncZZbl6Zngd/+Y+A6sgQ=
+github.com/libp2p/go-libp2p-core v0.5.7/go.mod h1:txwbVEhHEXikXn9gfC7/UDDw7rkxuX0bJvM49Ykaswo=
+github.com/libp2p/go-libp2p-testing v0.1.1 h1:U03z3HnGI7Ni8Xx6ONVZvUFOAzWYmolWf5W5jAOPNmU=
+github.com/libp2p/go-libp2p-testing v0.1.1/go.mod h1:xaZWMJrPUM5GlDBxCeGUi7kI4eqnjVyavGroI2nxEM0=
+github.com/libp2p/go-msgio v0.0.4/go.mod h1:63lBBgOTDKQL6EWazRMCwXsEeEeK9O2Cd+0+6OOuipQ=
+github.com/libp2p/go-openssl v0.0.5 h1:pQkejVhF0xp08D4CQUcw8t+BFJeXowja6RVcb5p++EA=
+github.com/libp2p/go-openssl v0.0.5/go.mod h1:unDrJpgy3oFr+rqXsarWifmJuNnJR4chtO1HmaZjggc=
+github.com/mailru/easyjson v0.0.0-20180823135443-60711f1a8329/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
+github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
+github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
+github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b/go.mod h1:01TrycV0kFyexm33Z7vhZRXopbI8J3TDReVlkTgMUxE=
+github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1 h1:lYpkrQH5ajf0OXOcUbGjvZxxijuBwbbmlSxLiuofa+g=
+github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1/go.mod h1:pD8RvIylQ358TN4wwqatJ8rNavkEINozVn9DtGI3dfQ=
+github.com/minio/sha256-simd v0.0.0-20190131020904-2d45a736cd16/go.mod h1:2FMWW+8GMoPweT6+pI63m9YE3Lmw4J71hV56Chs1E/U=
+github.com/minio/sha256-simd v0.0.0-20190328051042-05b4dd3047e5/go.mod h1:2FMWW+8GMoPweT6+pI63m9YE3Lmw4J71hV56Chs1E/U=
+github.com/minio/sha256-simd v0.1.0/go.mod h1:2FMWW+8GMoPweT6+pI63m9YE3Lmw4J71hV56Chs1E/U=
+github.com/minio/sha256-simd v0.1.1-0.20190913151208-6de447530771/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM=
+github.com/minio/sha256-simd v0.1.1 h1:5QHSlgo3nt5yKOJrC7W8w7X+NFl8cMPZm96iu8kKUJU=
+github.com/minio/sha256-simd v0.1.1/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM=
+github.com/mr-tron/base58 v1.1.0/go.mod h1:xcD2VGqlgYjBdcBLw+TuYLr8afG+Hj8g2eTVqeSzSU8=
+github.com/mr-tron/base58 v1.1.2/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
+github.com/mr-tron/base58 v1.1.3 h1:v+sk57XuaCKGXpWtVBX8YJzO7hMGx4Aajh4TQbdEFdc=
+github.com/mr-tron/base58 v1.1.3/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
+github.com/multiformats/go-base32 v0.0.3 h1:tw5+NhuwaOjJCC5Pp82QuXbrmLzWg7uxlMFp8Nq/kkI=
+github.com/multiformats/go-base32 v0.0.3/go.mod h1:pLiuGC8y0QR3Ue4Zug5UzK9LjgbkL8NSQj0zQ5Nz/AA=
+github.com/multiformats/go-multiaddr v0.0.4/go.mod h1:xKVEak1K9cS1VdmPZW3LSIb6lgmoS58qz/pzqmAxV44=
+github.com/multiformats/go-multiaddr v0.2.2 h1:XZLDTszBIJe6m0zF6ITBrEcZR73OPUhCBBS9rYAuUzI=
+github.com/multiformats/go-multiaddr v0.2.2/go.mod h1:NtfXiOtHvghW9KojvtySjH5y0u0xW5UouOmQQrn6a3Y=
+github.com/multiformats/go-multibase v0.0.1 h1:PN9/v21eLywrFWdFNsFKaU04kLJzuYzmrJR+ubhT9qA=
+github.com/multiformats/go-multibase v0.0.1/go.mod h1:bja2MqRZ3ggyXtZSEDKpl0uO/gviWFaSteVbWT51qgs=
+github.com/multiformats/go-multihash v0.0.1/go.mod h1:w/5tugSrLEbWqlcgJabL3oHFKTwfvkofsjW2Qa1ct4U=
+github.com/multiformats/go-multihash v0.0.5/go.mod h1:lt/HCbqlQwlPBz7lv0sQCdtfcMtlJvakRUn/0Ual8po=
+github.com/multiformats/go-multihash v0.0.13 h1:06x+mk/zj1FoMsgNejLpy6QTvJqlSt/BhLEy87zidlc=
+github.com/multiformats/go-multihash v0.0.13/go.mod h1:VdAWLKTwram9oKAatUcLxBNUjdtcVwxObEQBtRfuyjc=
+github.com/multiformats/go-varint v0.0.5 h1:XVZwSo04Cs3j/jS0uAEPpT3JY6DzMcVLLoWOSnCxOjg=
+github.com/multiformats/go-varint v0.0.5/go.mod h1:3Ls8CIEsrijN6+B7PbrXRPxHRPuXSrVKRY101jdMZYE=
+github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
+github.com/smola/gocompat v0.2.0/go.mod h1:1B0MlxbmoZNo3h8guHp8HztB3BSYR5itql9qtVc0ypY=
+github.com/spacemonkeygo/openssl v0.0.0-20181017203307-c2dcc5cca94a/go.mod h1:7AyxJNCJ7SBZ1MfVQCWD6Uqo2oubI2Eq2y2eqf+A5r0=
+github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572 h1:RC6RW7j+1+HkWaX/Yh71Ee5ZHaHYt7ZP4sQgUrm6cDU=
+github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572/go.mod h1:w0SWMsp6j9O/dk4/ZpIhL+3CkG8ofA2vuv7k+ltqUMc=
+github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
+github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
+github.com/src-d/envconfig v1.0.0/go.mod h1:Q9YQZ7BKITldTBnoxsE5gOeB5y66RyPXeue/R4aaNBc=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=
+go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
+go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
+golang.org/x/crypto v0.0.0-20170930174604-9419663f5a44/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/crypto v0.0.0-20190211182817-74369b46fc67/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20190618222545-ea8f1a30c443 h1:IcSOAf4PyMp3U3XbIEj1/xJ2BjNN2jWv7JoyOsMxXUU=
+golang.org/x/crypto v0.0.0-20190618222545-ea8f1a30c443/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
+golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
+golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
+golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
+golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190219092855-153ac476189d/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190502145724-3ef323f4f1fd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190626221950-04f50cda93cb h1:fgwFCsaw9buMuxNd6+DQfAuSFqbNiQZpcgJQAgJsK6k=
+golang.org/x/sys v0.0.0-20190626221950-04f50cda93cb/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20181030221726-6c7e314b6563/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20181130052023-1c3d964395ce/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
+golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
+google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
+google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
+google.golang.org/genproto v0.0.0-20190307195333-5fe7a883aa19/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE=
+google.golang.org/genproto v0.0.0-20190425155659-357c62f0e4bb/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE=
+google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
+google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
+gopkg.in/src-d/go-cli.v0 v0.0.0-20181105080154-d492247bbc0d/go.mod h1:z+K8VcOYVYcSwSjGebuDL6176A1XskgbtNl64NSg+n8=
+gopkg.in/src-d/go-log.v1 v1.0.1/go.mod h1:GN34hKP0g305ysm2/hctJ0Y8nWP3zxXXJ8GFabTyABE=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
+gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

opts.go

+package eventbus
+
+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
+		return nil
+	}
+}
+
+type emitterSettings struct {
+	makeStateful bool
+}
+
+// Stateful is an Emitter option which 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 interface{}) error {
+	s.(*emitterSettings).makeStateful = true
+	return nil
+}