EventBus

Package EventBus is the little and lightweight eventbus with async compatibility for GoLang.

Installation

Make sure that Go is installed on your computer. Type the following command in your terminal:

go get github.com/asaskevich/EventBus

After it the package is ready to use.

Import package in your project

Add following line in your *.go file:

import "github.com/asaskevich/EventBus"

If you unhappy to use long EventBus, you can do something like this:

import (
	evbus "github.com/asaskevich/EventBus"
)

Example

func calculator(a int, b int) {
	fmt.Printf("%d\n", a + b)
}

func main() {
	bus := EventBus.New();
	bus.Subscribe("main:calculator", calculator);
	bus.Publish("main:calculator", 20, 40);
	bus.Unsubscribe("main:calculator", calculator);
}

Implemented methods

• New()
• Subscribe()
• SubscribeOnce()
• HasCallback()
• Unsubscribe()
• Publish()
• SubscribeAsync()
• SubscribeOnceAsync()
• WaitAsync()

New()

New returns new EventBus with empty handlers.

bus := EventBus.New();

You can alternatively use ~50% faster implementation if your listener accepts only one argument of a particular type. The below example creates a new EventBus with handlers accepting only string arguments:

bus := EventBus.NewSimpleBus[string]();

Subscribe(topic string, fn interface{}) error

Subscribe to a topic. Returns error if fn is not a function.

func Handler() { ... }
...
bus.Subscribe("topic:handler", Handler)

SubscribeOnce(topic string, fn interface{}) error

Subscribe to a topic once. Handler will be removed after executing. Returns error if fn is not a function.

func HelloWorld() { ... }
...
bus.SubscribeOnce("topic:handler", HelloWorld)

Unsubscribe(topic string, fn interface{}) error

Remove callback defined for a topic. Returns error if there are no callbacks subscribed to the topic.

bus.Unsubscribe("topic:handler", HelloWord);

HasCallback(topic string) bool

Returns true if exists any callback subscribed to the topic.

Publish(topic string, args ...interface{})

Publish executes callback defined for a topic. Any additional argument will be transferred to the callback.

func Handler(str string) { ... }
...
bus.Subscribe("topic:handler", Handler)
...
bus.Publish("topic:handler", "Hello, World!");

In the case of simplified bus, there is only one argument accepted of predefined type.

SubscribeAsync(topic string, fn interface{}, transactional bool)

Subscribe to a topic with an asynchronous callback. Returns error if fn is not a function.

func slowCalculator(a, b int) {
	time.Sleep(3 * time.Second)
	fmt.Printf("%d\n", a + b)
}

bus := EventBus.New()
bus.SubscribeAsync("main:slow_calculator", slowCalculator, false)

bus.Publish("main:slow_calculator", 20, 60)

fmt.Println("start: do some stuff while waiting for a result")
fmt.Println("end: do some stuff while waiting for a result")

bus.WaitAsync() // wait for all async callbacks to complete

fmt.Println("do some stuff after waiting for result")

Transactional determines whether subsequent callbacks for a topic are run serially (true) or concurrently(false)

SubscribeOnceAsync(topic string, args ...interface{})

SubscribeOnceAsync works like SubscribeOnce except the callback to executed asynchronously

WaitAsync()

WaitAsync waits for all async callbacks to complete.

Cross Process Events

Works with two rpc services:

• a client service to listen to remotely published events from a server
• a server service to listen to client subscriptions

server.go

func main() {
    server := NewServer(":2010", "/_server_bus_", New())
    server.Start()
    // ...
    server.EventBus().Publish("main:calculator", 4, 6)
    // ...
    server.Stop()
}

client.go

func main() {
    client := NewClient(":2015", "/_client_bus_", New())
    client.Start()
    client.Subscribe("main:calculator", calculator, ":2010", "/_server_bus_")
    // ...
    client.Stop()
}

Benchmarks

% GOMAXPROCS=1 go test -bench=.
2
goos: darwin
goarch: amd64
pkg: github.com/ziollek/EventBus
cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz
BenchmarkSynchronousPublishing        	 2877297	       406.9 ns/op
BenchmarkAsynchronousPublishing       	  548457	      2101 ns/op
BenchmarkSimpleSynchronousPublishing  	 3939492	       305.5 ns/op
BenchmarkSimpleAsynchronousPublishing 	 1000000	      1141 ns/op
PASS
ok  	github.com/ziollek/EventBus	9.449s

% GOMAXPROCS=2 go test -bench=.
2
goos: darwin
goarch: amd64
pkg: github.com/ziollek/EventBus
cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz
BenchmarkSynchronousPublishing-2          	 3146671	       379.3 ns/op
BenchmarkAsynchronousPublishing-2         	 1050337	      1148 ns/op
BenchmarkSimpleSynchronousPublishing-2    	 5079916	       246.0 ns/op
BenchmarkSimpleAsynchronousPublishing-2   	 2124786	       548.2 ns/op
PASS
ok  	github.com/ziollek/EventBus	9.677s

Notes

Documentation is available here: godoc.org. Full information about code coverage is also available here: EventBus on gocover.io.

Support

If you do have a contribution for the package feel free to put up a Pull Request or open Issue.

Special thanks to contributors

• Brian Downs
• Dominik Schulz
• bennAH
• [John Noble] (https://github.com/gaxunil)
• [Evan Borgstrom] (https://github.com/borgstrom)
• ziollek