minibus

README

Minibus

Minibus is a very small in-memory message bus for Go.

Minibus executes a set of functions concurrently and exchanges messages between them. You can think of it like an errgroup.Group with a built-in message bus.

Example

A more detailed version of this example annotated with comments is available here.

type SayHello struct {
    Name string
}

minibus.Run(
    context.Background(),
    minibus.WithFunc(
        func(ctx context.Context) error {
            minibus.Subscribe[SayHello](ctx)
            minibus.Ready(ctx)

            for m := range minibus.Inbox(ctx) {
                switch m := m.(type) {
                case SayHello:
                    fmt.Printf("Hello, %s!\n", m.Name)
                }
            }

            return nil
        },
    ),
    minibus.WithFunc(
        func(ctx context.Context) error {
            minibus.Ready(ctx)
            return minibus.Send(ctx, SayHello{"world"})
        },
    ),
)

Documentation

Overview

Package minibus is a very small in-memory message bus.

Example

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/dogmatiq/minibus"
)

func main() {
	// SayHello is an example message type. Minibus doesn't care what types you
	// use for messages, but it's typical to use struct types.
	type SayHello struct {
		Name string
	}

	// The recipient function handles SayHello messages.
	recipient := func(ctx context.Context) error {
		// Subscribe to the types of messages we're interested in.
		minibus.Subscribe[SayHello](ctx)

		// All functions must signal readiness before messages are exchanged.
		minibus.Ready(ctx)

		// Handle the messages received on the function's inbox channel. It will
		// only receive messages of the same type that it subscribed to.
		//
		// The inbox channel is closed when ctx.Done() is closed, so it's not
		// necessary to select on both.
		for m := range minibus.Inbox(ctx) {
			switch m := m.(type) {
			case SayHello:
				fmt.Printf("Hello, %s!\n", m.Name)

				// We've said our greetings, let's get out of here.
				return nil
			}
		}

		// If the inbox channel was closed before we received the message it
		// means we were signalled to stop.
		return nil
	}

	// The sender function sends a SayHello message to the other functions.
	sender := func(ctx context.Context) error {
		minibus.Ready(ctx)
		return minibus.Send(ctx, SayHello{"world"})
	}

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	// The Run function executes each function in its own goroutine and
	// exchanges messages between them. It blocks until all functions return.
	if err := minibus.Run(
		ctx,
		recipient,
		sender,
	); err != nil {
		fmt.Println(err)
	}

}

Output:
Hello, world!

Index

• func Inbox(ctx context.Context) <-chan any
• func Outbox(ctx context.Context) chan<- any
• func Ready(ctx context.Context)
• func Receive(ctx context.Context) (any, error)
• func Run(ctx context.Context, functions ...Func) (err error)
• func Send(ctx context.Context, m any) error
• func Subscribe[M any](ctx context.Context)
• type Func
  • func Ingest[T any](messages <-chan T) Func

Examples

• Package
• Subscribe (FireHose)

Functions

func Inbox

func Inbox(ctx context.Context) <-chan any

Inbox returns the channel on which the function receives messages send by other functions executed by the same call to Run.

Only messages with types matching those passed to Subscribe will be received.

No messages are delivered until all functions executed by the same call to Run have called Ready.

func Outbox

func Outbox(ctx context.Context) chan<- any

Outbox returns a channel on which the function can send messages to other functions executed by the same call to Run.

The channel will block until all functions executed by the same call to Run have called Ready.

func Ready

func Ready(ctx context.Context)

Ready signals that the function has made all relevant Subscribe calls and is ready to exchange messages.

No messages are exchanged until all functions executed by the same call to Run have called Ready.

func Receive

func Receive(ctx context.Context) (any, error)

Receive returns the next received message, or an error if ctx is canceled.

func Run

func Run(
	ctx context.Context,
	functions ...Func,
) (err error)

Run exchanges messages between functions that it executes in parallel.

It blocks until all functions have returned, any single function returns an error, or ctx is canceled. Functions are added using the [WithFunc] option.

func Send

func Send(ctx context.Context, m any) error

Send sends a message, or returns an error if ctx is canceled.

func Subscribe

func Subscribe[M any](ctx context.Context)

Subscribe configures the calling function to receive messages of type M in its inbox.

It may only be called within a function that has been called by Run. It must be called before Ready.

Example (FireHose)

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/dogmatiq/minibus"
)

func main() {
	recipient := func(ctx context.Context) error {
		// Receive everything by subscribing to the [any] interface.
		minibus.Subscribe[any](ctx)
		minibus.Ready(ctx)

		for m := range minibus.Inbox(ctx) {
			fmt.Println(m)
		}

		return nil
	}

	sender := func(ctx context.Context) error {
		minibus.Ready(ctx)

		if err := minibus.Send(ctx, "Hello, world!"); err != nil {
			return err
		}

		return minibus.Send(ctx, 42)
	}

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

	if err := minibus.Run(
		ctx,
		recipient,
		sender,
	); err != context.DeadlineExceeded {
		fmt.Println(err)
	}

}

Output:
Hello, world!
42

Types

type Func

type Func func(context.Context) error

Func is a function that can be executed by Run.

func Ingest

func Ingest[T any](messages <-chan T) Func

Ingest returns a Func that reads messages from the provided channel and forwards them to the bus.