The #1 Go package for implementing the publish-subscribe pattern

Publish any type of data to any number of subscribers.

Report Bugs · Request Features · Twitter

About

This package provides a simple and efficient way to implement a publish/subscribe pattern in Go, allowing one sender to broadcast messages to multiple receivers. It supports dynamic subscription and unsubscription, ensuring flexibility and control over message delivery and resource management.

Features

• Generic Implementation: Works with any data type.
• Timeout Control: Customize the time to wait for subscribers to receive messages.
• Dynamic Subscription Management: Subscribers can join and leave at any time.
• Automatic Cleanup: Automatically closes all subscriber channels when the broadcaster is closed.

Getting Started

Installation

To use the broadcast library in your project, first, ensure your project is initialized as a Go module, then add the library to your project with:

go get github.com/njern/broadcast

Basic Usage

Start by importing the package in your Go file.

import (
    "github.com/njern/broadcast"
)

Create a broadcaster with a specified buffer size and timeout for broadcast operations. Use a timeout of 0 to wait indefinitely for each subscriber.

b := broadcast.New[string](10, time.Second)

Subscribers can now subscribe to the broadcaster, receiving a channel to listen for messages. The caller provides the channel buffer size as input (or 0 if you prefer an unbuffered channel).

ch, err := b.Subscribe(2)
if err != nil {
    log.Fatalf("Failed to subscribe: %v", err)
}

go func() {
    for msg := range ch {
        fmt.Println("Received:", msg)
    }
}()

Send messages to all subscribers by sending them to the broadcaster's channel.

b.Chan() <- "Hello, Broadcasters!"

Subscribers can stop receiving messages by unsubscribing.

b.Unsubscribe(ch)

When the broadcaster is no longer needed, close it to release all resources.

b.Close()

Custom Timeouts

You can control how long the broadcaster waits for subscribers to receive messages. This is set when creating the broadcaster and applies to all messages.

// This broadcaster will wait up to 10 seconds for subscribers to  
// receive a message before continuing to the next subscriber.
b := broadcast.New[int](10, 10*time.Second)

To disable timeouts and block until each subscriber receives a message (or the broadcaster is closed), pass 0.

b := broadcast.New[int](10, 0)

Handling Closed Broadcasters

Attempting to subscribe to a closed broadcaster will result in an ErrBroadcasterClosed error.

_, err := b.Subscribe(0)
if err == broadcast.ErrBroadcasterClosed {
    fmt.Println("Broadcaster has been closed.")
}

Contributing

Contributions to improve this library are welcome. Feel free to fork the repository, make your changes, and submit a pull request.

License

This library is licensed under the MIT License.

Acknowledgements

• I used the tips from this very neat article by the Daytona team to put together the README.