Building ch.Broadcast: One Event, Multiple Consumers in Go
In the previous posts, we built small channel helpers instead of jumping directly into larger abstractions.
ch.Send made channel sends context-aware.
ch.Receive made channel receives context-aware.
ch.Read made it possible to keep reading from a channel until either the input closes or the context is cancelled.
Now we can build another helper on top of them:
ch.Broadcast
The goal sounds simple:
Take one input channel and deliver every value to multiple output channels.
But the useful part is not the final helper.
The useful part is how the helper appears from a real coordination problem.
Start with a Real Event
Imagine an order is successfully created in an application.
At that moment, the system has one important event:
type OrderCreated struct {
OrderID string
UserID string
Total int64
Currency string
}
When an order is created, multiple things may need to happen:
- send a confirmation email
- update inventory
- notify the warehouse
- collect analytics
- write an audit log
A first version may look like this:
func createOrder(w http.ResponseWriter, r *http.Request) {
orderCreated := OrderCreated{
OrderID: "ord_123",
UserID: "user_456",
Total: 1299,
Currency: "TRY",
}
sendConfirmationEmail(orderCreated)
updateInventory(orderCreated)
notifyWarehouse(orderCreated)
trackAnalytics(orderCreated)
writeAuditLog(orderCreated)
w.WriteHeader(http.StatusCreated)
}
This works, but the handler now knows too much.
It knows every consumer of the event.
It knows how many consumers exist.
It knows which side effects must happen after an order is created.
That is not really the handler’s job.
The handler should create the order and publish the event.
The rest of the system should react to that event independently.
Publishing the Event
Let’s introduce a channel for order events:
orders := make(chan OrderCreated, 64)
Now the handler can publish an event after the order is created:
func createOrder(w http.ResponseWriter, r *http.Request) {
orderCreated := OrderCreated{
OrderID: "ord_123",
UserID: "user_456",
Total: 1299,
Currency: "TRY",
}
if err := ch.Send(r.Context(), orders, orderCreated); err != nil {
http.Error(w, "failed to publish order event", http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusCreated)
}
This keeps the example small, but do not ignore this failure in a real ordering system.
If the order has already been persisted and event publication fails, the system needs a recovery strategy.
That may mean marking the order as failed, retrying, or using the outbox pattern.
For this article, we will keep the example in memory so we can focus on the channel helper.
The important shift is this:
The handler only publishes OrderCreated.
It does not send emails.
It does not update inventory.
It does not notify the warehouse.
But now we have a new problem.
Multiple consumers need to receive the same event.
This Is Not Fan-Out
A common mistake is to put multiple goroutines on the same channel and call it broadcast.
For example:
go handleEmails(orders)
go updateInventory(orders)
go notifyWarehouse(orders)
This is not broadcast.
This is fan-out.
With fan-out, multiple consumers share the work.
One event goes to one consumer.
order-1 -> email worker
order-2 -> inventory worker
order-3 -> warehouse worker
That is useful, but it is not what we need here.
For an OrderCreated event, every consumer needs to see the same value.
order-1 -> email
order-1 -> inventory
order-1 -> warehouse
order-1 -> analytics
order-1 -> audit
That is broadcast.
One input.
Many outputs.
Same value delivered to all of them.
First Manual Broadcast
Let’s create separate channels for each consumer:
emailOrders := make(chan OrderCreated, 16)
inventoryOrders := make(chan OrderCreated, 16)
warehouseOrders := make(chan OrderCreated, 16)
analyticsOrders := make(chan OrderCreated, 16)
auditOrders := make(chan OrderCreated, 16)
Then we can manually copy each incoming order event into every output channel:
go func() {
for order := range orders {
emailOrders <- order
inventoryOrders <- order
warehouseOrders <- order
analyticsOrders <- order
auditOrders <- order
}
}()
Now every consumer can have its own channel:
go handleEmails(emailOrders)
go updateInventory(inventoryOrders)
go notifyWarehouse(warehouseOrders)
go trackAnalytics(analyticsOrders)
go writeAuditLog(auditOrders)
This is the first real version of broadcast.
The input is orders.
The outputs are consumer-specific channels.
Every order is copied to every output.
orders
|
v
broadcast loop
|
+--> emailOrders
+--> inventoryOrders
+--> warehouseOrders
+--> analyticsOrders
+--> auditOrders
So far, the idea is simple.
But the implementation is still fragile.
The Blocking Problem
This line can block:
inventoryOrders <- order
That is normal Go channel behavior.
If the channel is unbuffered, the send waits until a receiver is ready.
If the channel is buffered, the send waits when the buffer is full.
So if one consumer stops reading, the broadcast loop can get stuck.
orders
|
v
broadcast loop
|
+--> emailOrders ok
+--> inventoryOrders blocked
+--> warehouseOrders never reached
+--> analyticsOrders never reached
+--> auditOrders never reached
This is important.
A slow consumer can apply backpressure to the whole broadcast pipeline.
That is not always wrong.
For important events, blocking can be better than silently dropping data.
But the blocking should still be cancellable.
If the application is shutting down, the goroutine should not wait forever.
Adding Context-Aware Sends
Instead of raw channel sends:
inventoryOrders <- order
we can use the helper we already built:
if err := ch.Send(ctx, inventoryOrders, order); err != nil {
return
}
Now each send can stop when the context is cancelled.
The manual broadcast loop becomes:
go func() {
for order := range orders {
if err := ch.Send(ctx, emailOrders, order); err != nil {
return
}
if err := ch.Send(ctx, inventoryOrders, order); err != nil {
return
}
if err := ch.Send(ctx, warehouseOrders, order); err != nil {
return
}
if err := ch.Send(ctx, analyticsOrders, order); err != nil {
return
}
if err := ch.Send(ctx, auditOrders, order); err != nil {
return
}
}
}()
This is safer.
But now the problem is obvious.
We are not writing business logic anymore.
We are writing channel coordination logic.
The same pattern repeats for every output:
if err := ch.Send(ctx, out, value); err != nil {
return
}
That repetition is the first sign that a helper is trying to appear.
The Input Has the Same Problem
The loop still reads like this:
for order := range orders {
// send to outputs
}
This only stops when orders is closed.
But what if the context is cancelled before the input channel is closed?
That is exactly why ch.Read exists.
Instead of ranging over the raw input channel:
for order := range orders {
}
we range over a context-aware view of it:
for order := range ch.Read(ctx, orders) {
}
ch.Read keeps yielding values from the input channel until either:
- the input channel is closed
- the context is cancelled
Now our manual broadcast loop becomes:
go func() {
for order := range ch.Read(ctx, orders) {
if err := ch.Send(ctx, emailOrders, order); err != nil {
return
}
if err := ch.Send(ctx, inventoryOrders, order); err != nil {
return
}
if err := ch.Send(ctx, warehouseOrders, order); err != nil {
return
}
if err := ch.Send(ctx, analyticsOrders, order); err != nil {
return
}
if err := ch.Send(ctx, auditOrders, order); err != nil {
return
}
}
}()
This version is much closer to what we want.
But it still has a hard-coded list of outputs.
Too Many Outputs
Hard-coding every output does not scale well.
We could use a slice:
outputs := []chan OrderCreated{
emailOrders,
inventoryOrders,
warehouseOrders,
analyticsOrders,
auditOrders,
}
Then the loop becomes:
go func() {
for order := range ch.Read(ctx, orders) {
for _, out := range outputs {
if err := ch.Send(ctx, out, order); err != nil {
return
}
}
}
}()
This is much better.
But now the outputs are positional.
outputs[0]
outputs[1]
outputs[2]
What is outputs[0]?
Email?
Inventory?
Audit?
The code does not say.
For a low-level helper, a slice can be useful.
But for application wiring, named outputs are easier to understand.
So we can introduce a small type:
type Outputs[T any] map[string]chan T
Now the wiring is explicit:
outputs := ch.Outputs[OrderCreated]{
"email": make(chan OrderCreated, 16),
"inventory": make(chan OrderCreated, 16),
"warehouse": make(chan OrderCreated, 16),
"analytics": make(chan OrderCreated, 16),
"audit": make(chan OrderCreated, 16),
}
And the consumers are easy to read:
go handleEmails(outputs["email"])
go updateInventory(outputs["inventory"])
go notifyWarehouse(outputs["warehouse"])
go trackAnalytics(outputs["analytics"])
go writeAuditLog(outputs["audit"])
The names live in the application wiring.
ch.Broadcast does not need to know what email or inventory means.
It only needs to know:
Here are the output channels. Deliver every value to all of them.
Extracting ch.Broadcast
At this point, the pattern is clear.
We have:
- one input channel
- many output channels
- same value delivered to all outputs
- context-aware reading
- context-aware sending
- output channels closed when broadcasting ends
That is enough to extract a helper:
package ch
import "context"
type Outputs[T any] map[string]chan T
func Broadcast[T any](
ctx context.Context,
in <-chan T,
outs Outputs[T],
) {
defer func() {
for _, out := range outs {
close(out)
}
}()
for value := range Read(ctx, in) {
for _, out := range outs {
if err := Send(ctx, out, value); err != nil {
return
}
}
}
}
Now application code becomes much smaller:
outputs := ch.Outputs[OrderCreated]{
"email": make(chan OrderCreated, 16),
"inventory": make(chan OrderCreated, 16),
"warehouse": make(chan OrderCreated, 16),
"analytics": make(chan OrderCreated, 16),
"audit": make(chan OrderCreated, 16),
}
go ch.Broadcast(ctx, orders, outputs)
go handleEmails(outputs["email"])
go updateInventory(outputs["inventory"])
go notifyWarehouse(outputs["warehouse"])
go trackAnalytics(outputs["analytics"])
go writeAuditLog(outputs["audit"])
And the producer only publishes the event:
if err := ch.Send(r.Context(), orders, orderCreated); err != nil {
http.Error(w, "failed to publish order event", http.StatusInternalServerError)
return
}
Now let’s slow down and look at the implementation piece by piece.
The Outputs Type
type Outputs[T any] map[string]chan T
This type gives names to output channels.
The names are not used for routing decisions.
Broadcast sends every value to every output.
The names are for application readability.
This avoids code like:
outputs[0]
outputs[1]
outputs[2]
and gives us wiring like:
outputs["email"]
outputs["inventory"]
outputs["audit"]
The map values are bidirectional channels:
chan T
not receive-only channels:
<-chan T
because Broadcast needs to send values into them and close them when it is done.
Consumers still receive from them normally.
Closing the Output Channels
The first thing inside Broadcast is:
defer func() {
for _, out := range outs {
close(out)
}
}()
This means every output channel is closed when broadcasting stops.
That allows consumers to exit cleanly:
func handleEmails(in <-chan OrderCreated) {
for order := range in {
sendConfirmationEmail(order)
}
}
When the output channel is closed, the for range loop ends.
This follows the usual Go channel ownership rule:
The goroutine that sends values into a channel should be responsible for closing it.
In this design, Broadcast is the sender for all output channels.
So Broadcast closes them.
The consumers should not close these channels.
They only read from them.
Reading from the Input
The main loop starts with:
for value := range Read(ctx, in) {
This is intentionally not:
for value := range in {
A raw range in only stops when in is closed.
But broadcasting should also stop when the context is cancelled.
Read(ctx, in) gives us that behavior.
It creates a context-aware stream of values from the input channel.
So the broadcast loop continues while values are available and the context is still alive.
If the input closes, the loop ends.
If the context is cancelled, the loop ends.
That keeps the helper safe during shutdown.
Sending to Every Output
For each value, we iterate over all output channels:
for _, out := range outs {
if err := Send(ctx, out, value); err != nil {
return
}
}
This is the core of broadcast.
Every input value is sent to every output.
If there are five outputs, the same value is sent five times.
This is different from fan-out.
Fan-out distributes work across workers.
Broadcast duplicates visibility across consumers.
The Send call matters because a send may block.
For example, this can block:
out <- value
If the output channel buffer is full, or if the output channel is unbuffered and no receiver is ready, the send waits.
Send(ctx, out, value) makes that wait cancellable.
If the context is cancelled, Send returns and Broadcast exits.
What Happens When One Consumer Is Slow?
Suppose analytics stops reading.
orders
|
v
Broadcast
|
+--> email ok
+--> inventory ok
+--> warehouse ok
+--> analytics blocked
+--> audit waiting
Eventually, the analytics channel buffer may fill up.
When that happens, this send blocks:
Send(ctx, outputs["analytics"], value)
That means the whole broadcast loop waits.
This is intentional in the strict version.
ch.Broadcast does not silently drop events.
If all consumers are supposed to receive every value, then a slow consumer should create backpressure.
Buffering gives consumers some room to be temporarily slower.
But buffering does not remove backpressure forever.
That is usually a good default for important application events.
Buffering Is a Consumer Decision
Each output channel can choose its own buffer size:
outputs := ch.Outputs[OrderCreated]{
"email": make(chan OrderCreated, 16),
"inventory": make(chan OrderCreated, 32),
"warehouse": make(chan OrderCreated, 16),
"analytics": make(chan OrderCreated, 128),
"audit": make(chan OrderCreated, 64),
}
This is better than putting a single buffer int inside Broadcast.
Different consumers may have different characteristics.
Analytics may tolerate bursts.
Inventory may need tighter backpressure.
Audit may need enough room to absorb short spikes.
The buffer belongs to the channel.
So the caller creates the channels with the buffer sizes it wants.
Broadcast only coordinates delivery.
Where Broadcast Runs
Broadcast is usually started during application wiring:
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
orders := make(chan OrderCreated, 64)
outputs := ch.Outputs[OrderCreated]{
"email": make(chan OrderCreated, 16),
"inventory": make(chan OrderCreated, 16),
"warehouse": make(chan OrderCreated, 16),
"analytics": make(chan OrderCreated, 16),
"audit": make(chan OrderCreated, 16),
}
go ch.Broadcast(ctx, orders, outputs)
go handleEmails(outputs["email"])
go updateInventory(outputs["inventory"])
go notifyWarehouse(outputs["warehouse"])
go trackAnalytics(outputs["analytics"])
go writeAuditLog(outputs["audit"])
http.HandleFunc("/orders", createOrderHandler(orders))
http.ListenAndServe(":8080", nil)
}
The handler receives the orders channel:
func createOrderHandler(
orders chan<- OrderCreated,
) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
orderCreated := OrderCreated{
OrderID: "ord_123",
UserID: "user_456",
Total: 1299,
Currency: "TRY",
}
if err := ch.Send(r.Context(), orders, orderCreated); err != nil {
http.Error(w, "failed to publish order event", http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusCreated)
}
}
This separation is the main point.
The handler produces an event.
The broadcast wiring decides who receives it.
The consumers process it independently.
What ch.Broadcast Is Not
ch.Broadcast is not a durable message broker.
It does not replace Kafka, NATS, RabbitMQ, Redis Streams, or an outbox table.
It is an in-process channel helper.
If the process crashes, in-memory events are gone.
That means it is useful for local coordination inside a Go process.
For critical business events that must survive restarts, you usually need durable storage or a proper message system.
The value of ch.Broadcast is smaller and more focused:
It keeps native Go channel coordination readable, cancellable, and reusable.
Final Version
Here is the full helper again:
package ch
import "context"
type Outputs[T any] map[string]chan T
func Broadcast[T any](
ctx context.Context,
in <-chan T,
outs Outputs[T],
) {
defer func() {
for _, out := range outs {
close(out)
}
}()
for value := range Read(ctx, in) {
for _, out := range outs {
if err := Send(ctx, out, value); err != nil {
return
}
}
}
}
And here is the shape it gives us:
orders
|
v
ch.Broadcast
|
+--> email
+--> inventory
+--> warehouse
+--> analytics
+--> audit
Merge brings many inputs into one output.
Broadcast takes one input into many outputs.
ch.Merge: many inputs -> one output
ch.Broadcast: one input -> many outputs
Both are small coordination helpers.
Neither tries to hide Go channels.
They simply make common channel patterns easier to express without spreading fragile coordination logic across the application.