release v0.3.0

sync: update CI config files

.github/workflows/automerge.yml

@@ -0,0 +1,11 @@
+# File managed by web3-bot. DO NOT EDIT.
+# See https://github.com/protocol/.github/ for details.
+
+name: Automerge
+on: [ pull_request ]
+
+jobs:
+  automerge:
+    uses: protocol/.github/.github/workflows/automerge.yml@master
+    with:
+      job: 'automerge'

.github/workflows/go-check.yml

@@ -0,0 +1,73 @@
+# File managed by web3-bot. DO NOT EDIT.
+# See https://github.com/protocol/.github/ for details.
+
+on: [push, pull_request]
+name: Go Checks
+
+jobs:
+  unit:
+    runs-on: ubuntu-latest
+    name: All
+    env:
+      RUNGOGENERATE: false
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: recursive
+      - uses: actions/setup-go@v2
+        with:
+          go-version: "1.18.x"
+      - name: Run repo-specific setup
+        uses: ./.github/actions/go-check-setup
+        if: hashFiles('./.github/actions/go-check-setup') != ''
+      - name: Read config
+        if: hashFiles('./.github/workflows/go-check-config.json') != ''
+        run: |
+          if jq -re .gogenerate ./.github/workflows/go-check-config.json; then
+            echo "RUNGOGENERATE=true" >> $GITHUB_ENV
+          fi
+      - name: Install staticcheck
+        run: go install honnef.co/go/tools/cmd/staticcheck@d7e217c1ff411395475b2971c0824e1e7cc1af98 # 2022.1 (v0.3.0)
+      - name: Check that go.mod is tidy
+        uses: protocol/multiple-go-modules@v1.2
+        with:
+          run: |
+            go mod tidy
+            if [[ -n $(git ls-files --other --exclude-standard --directory -- go.sum) ]]; then
+              echo "go.sum was added by go mod tidy"
+              exit 1
+            fi
+            git diff --exit-code -- go.sum go.mod
+      - name: gofmt
+        if: ${{ success() || failure() }} # run this step even if the previous one failed
+        run: |
+          out=$(gofmt -s -l .)
+          if [[ -n "$out" ]]; then
+            echo $out | awk '{print "::error file=" $0 ",line=0,col=0::File is not gofmt-ed."}'
+            exit 1
+          fi
+      - name: go vet
+        if: ${{ success() || failure() }} # run this step even if the previous one failed
+        uses: protocol/multiple-go-modules@v1.2
+        with:
+          run: go vet ./...
+      - name: staticcheck
+        if: ${{ success() || failure() }} # run this step even if the previous one failed
+        uses: protocol/multiple-go-modules@v1.2
+        with:
+          run: |
+            set -o pipefail
+            staticcheck ./... | sed -e 's@\(.*\)\.go@./\1.go@g'
+      - name: go generate
+        uses: protocol/multiple-go-modules@v1.2
+        if: (success() || failure()) && env.RUNGOGENERATE == 'true'
+        with:
+          run: |
+            git clean -fd # make sure there aren't untracked files / directories
+            go generate ./...
+            # check if go generate modified or added any files
+            if ! $(git add . && git diff-index HEAD --exit-code --quiet); then
+              echo "go generated caused changes to the repository:"
+              git status --short
+              exit 1
+            fi

.github/workflows/go-test.yml

@@ -0,0 +1,68 @@
+# File managed by web3-bot. DO NOT EDIT.
+# See https://github.com/protocol/.github/ for details.
+
+on: [push, pull_request]
+name: Go Test
+
+jobs:
+  unit:
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ "ubuntu", "windows", "macos" ]
+        go: [ "1.17.x", "1.18.x" ]
+    env:
+      COVERAGES: ""
+    runs-on: ${{ format('{0}-latest', matrix.os) }}
+    name: ${{ matrix.os }} (go ${{ matrix.go }})
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: recursive
+      - uses: actions/setup-go@v2
+        with:
+          go-version: ${{ matrix.go }}
+      - name: Go information
+        run: |
+          go version
+          go env
+      - name: Use msys2 on windows
+        if: ${{ matrix.os == 'windows' }}
+        shell: bash
+        # The executable for msys2 is also called bash.cmd
+        #   https://github.com/actions/virtual-environments/blob/main/images/win/Windows2019-Readme.md#shells
+        # If we prepend its location to the PATH
+        #   subsequent 'shell: bash' steps will use msys2 instead of gitbash
+        run: echo "C:/msys64/usr/bin" >> $GITHUB_PATH
+      - name: Run repo-specific setup
+        uses: ./.github/actions/go-test-setup
+        if: hashFiles('./.github/actions/go-test-setup') != ''
+      - name: Run tests
+        uses: protocol/multiple-go-modules@v1.2
+        with:
+          # Use -coverpkg=./..., so that we include cross-package coverage.
+          # If package ./A imports ./B, and ./A's tests also cover ./B,
+          # this means ./B's coverage will be significantly higher than 0%.
+          run: go test -v -coverprofile=module-coverage.txt -coverpkg=./... ./...
+      - name: Run tests (32 bit)
+        if: ${{ matrix.os != 'macos' }} # can't run 32 bit tests on OSX.
+        uses: protocol/multiple-go-modules@v1.2
+        env:
+          GOARCH: 386
+        with:
+          run: |
+            export "PATH=${{ env.PATH_386 }}:$PATH"
+            go test -v ./...
+      - name: Run tests with race detector
+        if: ${{ matrix.os == 'ubuntu' }} # speed things up. Windows and OSX VMs are slow
+        uses: protocol/multiple-go-modules@v1.2
+        with:
+          run: go test -v -race ./...
+      - name: Collect coverage files
+        shell: bash
+        run: echo "COVERAGES=$(find . -type f -name 'module-coverage.txt' | tr -s '\n' ',' | sed 's/,$//')" >> $GITHUB_ENV
+      - name: Upload coverage to Codecov
+        uses: codecov/codecov-action@f32b3a3741e1053eb607407145bc9619351dc93b # v2.1.0
+        with:
+          files: '${{ env.COVERAGES }}'
+          env_vars: OS=${{ matrix.os }}, GO=${{ matrix.go }}

.github/workflows/release-check.yml

@@ -0,0 +1,11 @@
+# File managed by web3-bot. DO NOT EDIT.
+# See https://github.com/protocol/.github/ for details.
+
+name: Release Checker
+on:
+  pull_request:
+    paths: [ 'version.json' ]
+
+jobs:
+  release-check:
+    uses: protocol/.github/.github/workflows/release-check.yml@master

.github/workflows/releaser.yml

@@ -0,0 +1,11 @@
+# File managed by web3-bot. DO NOT EDIT.
+# See https://github.com/protocol/.github/ for details.
+
+name: Releaser
+on:
+  push:
+    paths: [ 'version.json' ]
+
+jobs:
+  releaser:
+    uses: protocol/.github/.github/workflows/releaser.yml@master

.github/workflows/tagpush.yml

@@ -0,0 +1,12 @@
+# File managed by web3-bot. DO NOT EDIT.
+# See https://github.com/protocol/.github/ for details.
+
+name: Tag Push Checker
+on:
+  push:
+    tags:
+      - v*
+
+jobs:
+  releaser:
+    uses: protocol/.github/.github/workflows/tagpush.yml@master

.travis.yml

@@ -1,30 +0,0 @@
-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

README.md

@@ -1,3 +1,7 @@
+# DEPRECATION NOTICE
+
+This package has moved into go-libp2p as a sub-package, `github.com/libp2p/go-libp2p/p2p/host/eventbus`.
+
 # go-eventbus
 
 [![](https://img.shields.io/badge/made%20by-Protocol%20Labs-blue.svg?style=flat-square)](https://protocol.ai)

basic.go

@@ -1,263 +1,12 @@
+// Deprecated: This package has moved into go-libp2p as a sub-package: github.com/libp2p/go-libp2p/p2p/host/eventbus.
 package eventbus
 
 import (
-	"errors"
-	"fmt"
-	"reflect"
-	"sync"
-	"sync/atomic"
-
-	"github.com/libp2p/go-libp2p-core/event"
+	"github.com/libp2p/go-libp2p/core/event"
+	"github.com/libp2p/go-libp2p/p2p/host/eventbus"
 )
 
-///////////////////////
-// BUS
-
-// basicBus is a type-based event delivery system
-type basicBus struct {
-	lk    sync.Mutex
-	nodes map[reflect.Type]*node
-}
-
-var _ event.Bus = (*basicBus)(nil)
-
-type emitter struct {
-	n       *node
-	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)
-	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
-}
-
+// Deprecated: Use github.com/libp2p/go-libp2p/p2p/host/eventbus.NewBus instead.
 func NewBus() event.Bus {
-	return &basicBus{
-		nodes: map[reflect.Type]*node{},
-	}
-}
-
-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 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 := subSettings(subSettingsDefault)
-	for _, opt := range opts {
-		if err := opt(&settings); err != nil {
-			return nil, err
-		}
-	}
-
-	types, ok := evtTypes.([]interface{})
-	if !ok {
-		types = []interface{}{evtTypes}
-	}
-
-	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) {
-	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}
-	}, nil)
-	return
-}
-
-///////////////////////
-// NODE
-
-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(event interface{}) {
-	typ := reflect.TypeOf(event)
-	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 = event
-	}
-
-	for _, ch := range n.sinks {
-		ch <- event
-	}
-	n.lk.Unlock()
+	return eventbus.NewBus()
 }

basic_test.go

@@ -1,533 +0,0 @@
-package eventbus
-
-import (
-	"fmt"
-	"sync"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"github.com/libp2p/go-libp2p-testing/race"
-)
-
-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 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 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))
-		}
-	}
-}

go.mod

@@ -1,8 +1,29 @@
 module github.com/libp2p/go-eventbus
 
-go 1.12
+go 1.17
+
+require github.com/libp2p/go-libp2p v0.22.0
 
 require (
-	github.com/libp2p/go-libp2p-core v0.5.0
-	github.com/libp2p/go-libp2p-testing v0.1.1
+	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.1.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/ipfs/go-cid v0.2.0 // indirect
+	github.com/klauspost/cpuid/v2 v2.1.0 // indirect
+	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
+	github.com/libp2p/go-openssl v0.1.0 // indirect
+	github.com/mattn/go-pointer v0.0.1 // indirect
+	github.com/minio/sha256-simd v1.0.0 // indirect
+	github.com/mr-tron/base58 v1.2.0 // indirect
+	github.com/multiformats/go-base32 v0.0.4 // indirect
+	github.com/multiformats/go-base36 v0.1.0 // indirect
+	github.com/multiformats/go-multiaddr v0.6.0 // indirect
+	github.com/multiformats/go-multibase v0.1.1 // indirect
+	github.com/multiformats/go-multicodec v0.5.0 // indirect
+	github.com/multiformats/go-multihash v0.2.1 // indirect
+	github.com/multiformats/go-varint v0.0.6 // indirect
+	github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572 // indirect
+	github.com/spaolacci/murmur3 v1.1.0 // indirect
+	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e // indirect
+	golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab // indirect
+	lukechampine.com/blake3 v1.1.7 // indirect
 )

opts.go

@@ -1,32 +1,20 @@
 package eventbus
 
-type subSettings struct {
-	buffer int
-}
-
-var subSettingsDefault = subSettings{
-	buffer: 16,
-}
+import "github.com/libp2p/go-libp2p/p2p/host/eventbus"
 
+// Deprecated: Use github.com/libp2p/go-libp2p/p2p/host/eventbus.BufSize instead.
 func BufSize(n int) func(interface{}) error {
-	return func(s interface{}) error {
-		s.(*subSettings).buffer = n
-		return nil
-	}
+	return eventbus.BufSize(n)
 }
 
-type emitterSettings struct {
-	makeStateful bool
-}
-
-// Stateful is an Emitter option which makes makes the eventbus channel
+// 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
+// Deprecated: Use github.com/libp2p/go-libp2p/p2p/host/eventbus.Stateful instead.
 func Stateful(s interface{}) error {
-	s.(*emitterSettings).makeStateful = true
-	return nil
+	return eventbus.Stateful(s)
 }

version.json

@@ -0,0 +1,3 @@
+{
+  "version": "v0.3.0"
+}