Skip to content
master
Switch branches/tags
Code

Latest commit

 

Git stats

Files

Permalink
Failed to load latest commit information.
Type
Name
Latest commit message
Commit time
 
 
 
 
 
 
 
 
 
 
 
 

broadcast

CI Go Report Card

Notifications broadcaster in Go

What?

broadcast is a library that allows sending repeated notifications to multiple goroutines with guaranteed delivery.

Why?

Why not Channels?

The standard way to handle notifications is via a chan struct{}. However, sending a message to a channel is received by a single goroutine.

The only operation that is broadcasted to multiple goroutines is a channel closure. Yet, if the channel is closed, there's no way to send a message again.

Repeated notifications to multiple goroutines

Guaranteed delivery

Why not sync.Cond?

sync.Cond is the standard solution based on condition variables to set up containers of goroutines waiting for a specific condition.

There's one caveat to keep in mind, though: the Broadcast() method doesn't guarantee that a goroutine will receive the notification. Indeed, the notification will be lost if the listener goroutine isn't waiting on the Wait() method.

Repeated notifications to multiple goroutines

Guaranteed delivery

How?

Step by Step

First, we need to create a Relay:

relay := broadcast.NewRelay()

Once a Relay is created, we can create a new listener using the Listen method. As the broadcast library relies internally on channels, it accepts a capacity:

list := relay.Listener(1) // Create a new listener based on a channel with a one capacity

A Relay can send a notification in three different manners:

  • Notify(): block until a notification is sent to all the listeners
  • NotifyCtx(context.Context): send a notification to all the listeners unless the provided context times out or is canceled
  • Broadcast(): send a notification to all the listeners in a non-blocking manner; guaranteed delivery isn't guaranteed

On the Listener side, we can access the internal channel using Ch:

<-list.Ch() // Wait on a notification

We can close a Listener and a Relay using Close:

list.Close() 
relay.Close()

Closing a Relay and Listeners can be done concurrently in a safe manner.

Example

relay := broadcast.NewRelay() // Create a relay
defer relay.Close()

list1 := relay.Listener(1) // Create a listener with one capacity
defer list1.Close()
list2 := relay.Listener(1) // Create a listener with one capacity
defer list2.Close()

// Listener goroutines
f := func(i int, list *broadcast.Listener) {
    for range list.Ch() { // Waits for receiving notifications
        fmt.Printf("listener %d has received a notification\n", i)
    }
}
go f(1, list1)
go f(2, list2)

// Notifier goroutine
for i := 0; i < 5; i++ {
time.Sleep(time.Second)
    relay.Notify() // Send notifications with guaranteed delivery
    ctx, _ := context.WithTimeout(context.Background(), time.Second)
    relay.NotifyCtx(ctx) // Send notifications but cancel after 1 second
    relay.Broadcast()    // Send notifications without guaranteed delivery
}