Publishing

func (b *Bus[T]) Publish(ctx context.Context, topic string, msg T) error

Publish puts msg in the queue of every subscriber of topic, in the order they subscribed, and returns. It doesn’t wait for the handlers: they take the message from their queues on their own goroutines.

err := bus.Publish(ctx, "orders", OrderPlaced{ID: 7, Total: 1250})

A topic is any string, and needs no setup: it exists while it has subscribers. A message published to a topic nobody subscribes to is dropped, and Publish returns nil.

Order

Each subscriber gets its messages in the order its queue took them. So messages you publish from one goroutine arrive in the order you published them, at every subscriber.

Messages from goroutines publishing at the same time have no order between them, and can arrive in a different order at different subscribers. If two messages must stay in order, publish them from the same goroutine.

When a queue is full

Every subscriber has a queue of the size you gave New. A handler that falls behind fills it, and then Publish waits for room: until the handler takes a message, the subscriber unsubscribes, or ctx is done.

While it waits, Publish holds no lock. Publishing to other topics, subscribing, unsubscribing and closing all carry on; only this call waits, and the subscribers after the full one get the message once it fits.

ctx bounds the wait. When it is done, Publish returns its error, wrapped, and gives up on this message for the full subscriber and the ones after it; the ones before already have it.

ctx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
defer cancel()

err := bus.Publish(ctx, "jobs", job)
if errors.Is(err, context.DeadlineExceeded) {
  // a subscriber is behind: drop the job, or try again later
}

Publish looks at ctx only when it has to wait. A message that fits in every queue is queued even if ctx is already done.

Publishing from a handler

A handler that publishes to its own topic and finds its own queue full waits for room that only it can make. Give that Publish a context with a deadline, or the bus a bigger queue.

Unbuffered queues

messagebus.New[T](0) gives every subscriber a queue with no room: Publish hands each message straight to the handler, and waits whenever the handler is busy. Use it when the publisher should never get ahead of the slowest subscriber.

Errors

Error	When
nil	Every subscriber’s queue took the message, apart from any that unsubscribed while Publish waited for it. Also when the topic has no subscribers.
ctx’s error, wrapped	ctx was done while a queue was full. Check with errors.Is(err, context.DeadlineExceeded).
ErrClosed	Close was called before the call, or while it waited for room and before every queue had the message. The queues that took it still have it handled.

Example: skipping work when a handler falls behind

This program publishes jobs faster than its handler takes them. Each Publish waits at most 30ms for room, then skips the job:

examples/backpressure/main.go

// Command backpressure publishes jobs faster than its handler can take them. When the queue is full,
// it waits a little for room, then skips the job rather than stall.
//
//  go run ./examples/backpressure
package main

import (
  "context"
  "errors"
  "fmt"
  "log"
  "time"

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

// publishAll publishes every job, waiting at most wait for room each time, and returns the jobs that didn't fit.
func publishAll(ctx context.Context, bus *messagebus.Bus[int], jobs []int, wait time.Duration) ([]int, error) {
  var skipped []int
  for _, job := range jobs {
    err := publishWithin(ctx, bus, job, wait)
    if errors.Is(err, context.DeadlineExceeded) && ctx.Err() == nil {
      skipped = append(skipped, job)
      continue
    }
    if err != nil {
      return skipped, err
    }
  }
  return skipped, nil
}

func publishWithin(ctx context.Context, bus *messagebus.Bus[int], job int, wait time.Duration) error {
  ctx, cancel := context.WithTimeout(ctx, wait)
  defer cancel()
  return bus.Publish(ctx, "jobs", job)
}

func main() {
  ctx := context.Background()
  // The handler can hold one job and queue two more.
  bus := messagebus.New[int](2)
  if _, err := bus.Subscribe("jobs", func(job int) {
    time.Sleep(100 * time.Millisecond)
    fmt.Println("done", job)
  }); err != nil {
    log.Fatal(err)
  }

  skipped, err := publishAll(ctx, bus, []int{1, 2, 3, 4, 5, 6, 7, 8}, 30*time.Millisecond)
  if err != nil {
    log.Fatal(err)
  }
  fmt.Println("skipped", skipped)

  if err := bus.Close(ctx); err != nil {
    log.Fatal(err)
  }
}

done 1
skipped [4 5 6 8]
done 2
done 3
done 7