One message, every subscriber

Publish typed messages to topics within one Go program. Each subscriber handles them in order on a goroutine of its own, Publish waits for room without holding a lock, and Close waits until every message is handled.

Get started View on GitHub

go get github.com/vardius/message-bus/v2

// Every subscriber gets each order, in order,
// on a goroutine of its own.
bus := messagebus.New[OrderPlaced](100)

unsubscribe, err := bus.Subscribe("orders",
  func(e OrderPlaced) {
    sendReceipt(e)
  },
)
defer unsubscribe()

// Queues the order for every subscriber.
// A full queue makes it wait for room, or ctx.
err := bus.Publish(ctx, "orders",
  OrderPlaced{ID: 7, Total: 1250},
)
if errors.Is(err, context.DeadlineExceeded) {
  // a subscriber is falling behind
}

// Stop publishing, then wait until every
// queued message has been handled.
srv.Shutdown(ctx)

if err := bus.Close(ctx); err != nil {
  log.Printf("not all handled: %v", err)
}

3.9×faster than v1 to publish to a subscriber
5.7×faster than v1 with 16 goroutines publishing to 100 topics
0allocations to publish while there is room, however many subscribers
100%test coverage; the concurrency tests run in testing/synctest bubbles

Features

What you get

Typed messages

A Bus[T] carries one message type and handlers are func(T). No reflection, no interface{}: a handler that doesn’t fit is a compile error.

In order, in parallel

Each subscriber has its own queue and goroutine, and handles its messages one at a time, in the order they were queued.

Bounded waits

When a queue is full, Publish waits for room without holding a lock, and gives up when its context is done. The rest of the bus carries on.

Unsubscribe with a handle

Subscribe returns the func that ends that subscription, and only that one. A handler can call it to unsubscribe itself.

Graceful shutdown

Close waits until every queued message has been handled, or its context is done. A message Publish queued is never dropped.

Lean

Publish allocates nothing while there is room, and tests guard the allocation counts. No dependencies beyond the standard library.

Performance

Faster than v1, without an allocation

v2 calls handlers directly instead of through reflection, and spreads topics over 64 locks, so publishers to different topics rarely meet. Publishing to one subscriber went from 190.9 ns to 48.6 ns. With 16 publishers at once, the time is wall time per call across all of them, so it measures throughput.

Time per call to Publish, lower is better

Show as a table

Time per call to Publish: time and allocations, v1 against v2
Scenario	v1	v2
1 subscriber	190.9 ns2 allocs	48.6 ns0 allocs
10 subscribers	895.0 ns2 allocs	449.3 ns0 allocs
100 subscribers	13.02 µs2 allocs	9.81 µs0 allocs
100 topics, 1 subscriber each	146.8 ns2 allocs	112.2 ns0 allocs

Apple M4 Max, Go 1.27.0, median of 5 runs.

Time per call to Publish, 16 goroutines at once, lower is better

Show as a table

Time per call to Publish, 16 goroutines at once: time and allocations, v1 against v2
Scenario	v1	v2
1 subscriber	347.5 ns2 allocs	109.4 ns0 allocs
100 topics, 1 subscriber each	159.2 ns2 allocs	27.7 ns0 allocs

Apple M4 Max, Go 1.27.0, median of 5 runs.

Time per subscription, lower is better

Show as a table

Time per subscription: time and allocations, v1 against v2
Scenario	v1	v2
Subscribe, then unsubscribe	2.45 µs5 allocs	1.12 µs6 allocs

Apple M4 Max, Go 1.27.0, median of 5 runs.

How these were measuredMethodology, the full v2 suite, and how to run it yourself.

Upgrading

Coming from v1

The idea is the same: publish to a topic, and every subscriber gets the message. The bus is typed now, Subscribe hands back a func to unsubscribe with, and Publish takes a context.

bus := messagebus.New(100)

handler := func(e OrderPlaced) {
  sendReceipt(e)
}
bus.Subscribe("orders", handler)
bus.Publish("orders", OrderPlaced{ID: 7})

bus.Unsubscribe("orders", handler)

bus := messagebus.New[OrderPlaced](100)

unsubscribe, err := bus.Subscribe("orders",
  func(e OrderPlaced) {
    sendReceipt(e)
  },
)
err = bus.Publish(ctx, "orders",
  OrderPlaced{ID: 7})

unsubscribe()

Migration guideEvery v1 call and its v2 form, and the behaviour changes to know about.

Publish your first event in five minutes

Install the module, subscribe two handlers, and publish to both. The guide walks through full queues and shutdown too.

Get started Read the docs