mpe85 / go-broker

A generic thread-safe message broker for Go

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)
}

go-broker's Issues

Dependency Dashboard

This issue lists Renovate updates and detected dependencies. Read the Dependency Dashboard docs to learn more.

This repository currently has no open or pending branches.

Detected dependencies

github-actions

.github/workflows/test.yml

• actions/checkout v4
• actions/setup-go v5
• codecov/codecov-action v4

gomod

go.mod

• go 1.20
• github.com/stretchr/testify v1.9.0
• go.uber.org/goleak v1.3.0

---

• Check this box to trigger a request for Renovate to run again on this repository