go-broker

A generic thread-safe message broker for Go

This library provides a broker implementation that handles publishing of messages in a thread-safe manner. It supports multiple concurrent publishers as well as multiple clients subscribing to the broker. All operations on the broker (like publish, subscribe, unsubscribe) are synchronized, but may time out if the internal broker loop is too busy. The size of the internal buffer that buffers published messages is configurable, as well as the timeout duration.

Installation

go get github.com/mpe85/go-broker

Usage

Build a new broker with default configuration:

theBroker := broker.New[string]()

Build a new broker with custom configuration:

theBroker := broker.NewBuilder[string]().
	Timeout(3 * time.Second).
	BufferSize(100).
	Build()

Subscribe to the broker:

client, err := theBroker.Subscribe()

Unsubscribe from the broker:

err := theBroker.Unsubscribe(client)

Publish a message to the broker:

err := theBroker.Publish("Hello")

Receive a single message from the broker:

message := <-client

Receive a single message from the broker, with check if client is closed:

message, ok := <-client

Iterate over all messages from the broker, until client is closed:

for message := range client {
	// process message
}

Shutdown the broker, and close all clients that are still subscribed:

theBroker.Close()

Example

package main

import (
	"fmt"
	"time"

	"github.com/mpe85/go-broker"
)

func main() {
	theBroker := broker.NewBuilder[int]().
		Timeout(100 * time.Millisecond).
		BufferSize(50).
		Build()

	defer theBroker.Close()

	client1, _ := theBroker.Subscribe()
	client2, _ := theBroker.Subscribe()

	go func() {
		_ = theBroker.Publish(42)
	}()
	
	fmt.Println(<-client1)
	fmt.Println(<-client2)
}