package hub
type Hub struct {
subscribers map[*Subscriber]struct{}
broadcast chan []byte
register chan registration
unregister chan *Subscriber
done chan struct{}
}
type registration struct {
sub *Subscriber
reply chan struct{}
}
func New() *Hub {
return &Hub{
subscribers: make(map[*Subscriber]struct{}),
broadcast: make(chan []byte, 64),
register: make(chan registration),
unregister: make(chan *Subscriber),
done: make(chan struct{}),
}
}
func (h *Hub) Run() {
for {
select {
case reg := <-h.register:
h.subscribers[reg.sub] = struct{}{}
close(reg.reply)
case sub := <-h.unregister:
if _, ok := h.subscribers[sub]; ok {
delete(h.subscribers, sub)
close(sub.events)
}
case msg := <-h.broadcast:
for sub := range h.subscribers {
select {
case sub.events <- msg:
default:
// slow consumer: drop it to protect the rest
delete(h.subscribers, sub)
close(sub.events)
}
}
case <-h.done:
for sub := range h.subscribers {
close(sub.events)
}
return
}
}
}
func (h *Hub) Stop() {
close(h.done)
}package hub
type Subscriber struct {
hub *Hub
events chan []byte
}
func (h *Hub) Subscribe(buffer int) *Subscriber {
sub := &Subscriber{
hub: h,
events: make(chan []byte, buffer),
}
reply := make(chan struct{})
select {
case h.register <- registration{sub: sub, reply: reply}:
<-reply
case <-h.done:
}
return sub
}
func (s *Subscriber) Events() <-chan []byte {
return s.events
}
func (s *Subscriber) Close() {
select {
case s.hub.unregister <- s:
case <-s.hub.done:
}
}
func (h *Hub) Publish(msg []byte) bool {
select {
case h.broadcast <- msg:
return true
case <-h.done:
return false
}
}package main
import (
"fmt"
"sync"
"time"
"example.com/app/hub"
)
func main() {
h := hub.New()
go h.Run()
defer h.Stop()
var wg sync.WaitGroup
for i := 0; i < 3; i++ {
sub := h.Subscribe(8)
wg.Add(1)
go func(id int, s *hub.Subscriber) {
defer wg.Done()
for msg := range s.Events() {
fmt.Printf("subscriber %d received: %s\n", id, msg)
}
}(i, sub)
}
for n := 0; n < 5; n++ {
h.Publish([]byte(fmt.Sprintf("event #%d", n)))
time.Sleep(20 * time.Millisecond)
}
time.Sleep(100 * time.Millisecond)
h.Stop()
wg.Wait()
}This snippet implements a classic pub/sub fan-out hub in Go, where a single stream of events is broadcast to many independent subscribers. The pattern shows up behind WebSocket servers, live dashboards, and internal event buses — anywhere one producer must feed many consumers without them blocking each other.
In hub.go, the Hub owns all mutable state and confines it to a single goroutine running Run. Rather than guarding a subscriber map with a mutex, the hub uses the actor-style approach: registration, unregistration, and broadcasts all arrive as messages on channels (register, unregister, broadcast), and the select loop is the only place that touches subscribers. This eliminates data races by design because no two goroutines ever read or write the map concurrently.
The critical detail is the inner select in the broadcast case. Each subscriber has a small buffered channel, and the hub attempts a non-blocking send with a default branch. If a subscriber's buffer is full — a slow consumer exhibiting backpressure — the hub does not wait; instead it drops that subscriber and closes its channel. This protects the whole system from one lagging client stalling every other subscriber, a common failure mode in naive broadcast code that ranges over channels with plain blocking sends.
In subscriber.go, Subscriber wraps the receive channel and exposes Events as a read-only channel plus a Close method that signals the hub to remove it. The Publish method on the Hub is safe to call from any goroutine and itself uses a select against done so publishing never blocks forever during shutdown.
main.go wires it together: it starts the hub, registers a few subscribers, launches a publisher goroutine, and drains events. Note how Subscribe returns both the subscriber and participates in the register handshake via a reply channel, so the caller gets a fully-registered handle before any events flow.
The trade-off is deliberate: bounded buffers plus drop-on-full favors liveness and low latency over guaranteed delivery. When every message matters, a developer would instead add per-subscriber durable queues or acknowledgements. For fan-out of ephemeral events — presence, telemetry, notifications — dropping a slow subscriber is usually the correct, self-healing behavior.
Related snips
package api
import (
"net/http"
"runtime/debug"
)Expose build metadata for debugging deploys
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
