Getting started

message-bus delivers messages between the parts of one Go program. A part publishes a message to a topic; every part subscribed to that topic gets it, on a goroutine of its own, without the publisher knowing who they are.

Publish puts a message in the queue of every subscriber of a topic, and returns.
Subscribe runs a handler for every message on a topic, one at a time, in order.
Close waits until every queued message has been handled, then shuts the bus down.

Install

message-bus needs Go 1.26 or newer.

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

Your first bus

This program publishes three orders. One handler writes a receipt for each; another takes what was ordered out of stock. Neither knows about the other, and the code that publishes knows about neither:

// Command events publishes orders to two handlers that don't know about each other:
// one writes receipts, the other keeps the stock.
//
//  go run ./examples/events
package main

import (
  "context"
  "fmt"
  "io"
  "log"
  "os"

  messagebus "github.com/vardius/message-bus/v2"
)

type OrderPlaced struct {
  ID       int
  Item     string
  Quantity int
}

// placeOrders publishes every order, writes a receipt for each, and takes what was ordered out of stock.
func placeOrders(ctx context.Context, receipts io.Writer, stock map[string]int, orders []OrderPlaced) error {
  bus := messagebus.New[OrderPlaced](100)

  // Each handler runs on its own goroutine, one order at a time, so stock needs no lock:
  // only this handler touches it until Close returns.
  if _, err := bus.Subscribe("orders", func(o OrderPlaced) {
    stock[o.Item] -= o.Quantity
  }); err != nil {
    return err
  }
  if _, err := bus.Subscribe("orders", func(o OrderPlaced) {
    fmt.Fprintf(receipts, "order %d: %d × %s\n", o.ID, o.Quantity, o.Item)
  }); err != nil {
    return err
  }

  for _, o := range orders {
    if err := bus.Publish(ctx, "orders", o); err != nil {
      return err
    }
  }

  // Close returns once both handlers have handled every order.
  return bus.Close(ctx)
}

func main() {
  stock := map[string]int{"tea": 10, "cake": 4}
  orders := []OrderPlaced{
    {ID: 1, Item: "tea", Quantity: 2},
    {ID: 2, Item: "cake", Quantity: 1},
    {ID: 3, Item: "tea", Quantity: 3},
  }

  if err := placeOrders(context.Background(), os.Stdout, stock, orders); err != nil {
    log.Fatal(err)
  }
  fmt.Println("left:", stock)
}

Run it

From a clone of the repository:

go run ./examples/events

order 1: 2 × tea
order 2: 1 × cake
order 3: 3 × tea
left: map[cake:3 tea:5]

Add a fourth order to orders in main and run it again. The receipts stay in the order the orders were published, because each subscriber handles its messages one at a time.

What just happened

messagebus.New[OrderPlaced](100) made a bus for OrderPlaced messages. Each subscriber gets a queue of 100 of them.
Each Subscribe started a goroutine that waits for messages on "orders" and calls its handler with each one.
Publish copied the order into both queues and returned at once: it waits for room, never for handlers.
Close ended both subscriptions, and returned once both handlers had handled every order. Only then is stock safe to read from main.

Next steps

PublishingOrder, full queues, contexts and errors.

SubscribingHandlers, their goroutines, and unsubscribing.

ClosingShut down without losing a queued message.

BenchmarksWhat each call costs, and how v2 compares with v1.