minibus

README

Minibus

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

Documentation

Overview

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

Example

package main

import (
	"context"
	"fmt"

	"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 implements a "component" that receives SayHello
	// messages.
	recipient := func(ctx context.Context) error {
		// First, we subscribe to the types of messages we're interested in.
		minibus.Subscribe[SayHello](ctx)

		// All components must be started before we can send or receive
		// messages.
		minibus.Start(ctx)

		// Finally, we receive messages from our component-specific inbox.
		//
		// 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)
				return nil // let's not wait for any more greetings
			}
		}

		// If the inbox is closed before we received our message it means we
		// were signaled to stop.
		return ctx.Err()
	}

	// The sender function implements a "component" that sends a SayHello
	// message.
	sender := func(ctx context.Context) error {
		// Start must be called by all components, even if they do not subscribe
		// to any message types.
		minibus.Start(ctx)

		// Then we say hello to the other component!
		return minibus.Send(ctx, SayHello{"world"})
	}

	// The Run function executes each component in its own goroutine and
	// delivers messages between them. It blocks until all component's
	// goroutines have exited.
	if err := minibus.Run(
		context.Background(),
		minibus.WithComponent(recipient),
		minibus.WithComponent(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 Receive(ctx context.Context) (any, error)
• func Run(ctx context.Context, option ...RunOption) error
• func Send[M any](ctx context.Context, m M) error
• func Start(ctx context.Context)
• func Subscribe[M any](ctx context.Context)
• type RunOption
  • func WithBuffer(size int) RunOption
  • func WithComponent(run func(context.Context) error) RunOption

Examples

• Package

Functions

func Inbox

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

Inbox returns a channel that receives messages sent to the calling component.

No messages will be received until all components have started.

func Outbox

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

Outbox returns a channel that sends messages to other components.

func Receive

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

Receive is a convenience function for receiving the next message from the bus, or aborting if ctx is canceled.

func Run

func Run(ctx context.Context, option ...RunOption) error

Run starts a messaging session.

It blocks until all components (see WithComponent) have stopped, an error occurs, or ctx is canceled.

func Send

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

Send is a convenience function for sending a message to the bus, or aborting if ctx is canceled.

func Start

func Start(ctx context.Context)

Start starts the calling component.

No messages will be received until all components have started.

func Subscribe

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

Subscribe subscribes the calling component to receive messages of type M.

It must be called before Start.

Types

type RunOption

type RunOption func(*runOptions)

RunOption is a functional option for configuring the behavior of Run.

func WithBuffer

func WithBuffer(size int) RunOption

WithBuffer is an RunOption that sets the buffer size for each component's inbox channel.

The default buffer size is 10.

func WithComponent

func WithComponent(run func(context.Context) error) RunOption

WithComponent is a RunOption that adds an application component to the messaging session.

run is called in its own goroutine. It may use the Start, Inbox, Outbox, Send and Receive functions to perform messaging operations.