Observer Pattern — Hands-on Tasks¶

Work through these in order. The first few drill the basic Subject/Observer shape — a subject that holds a list of observers and calls Update on each when something changes. The middle tasks force the Go-specific design decisions: channels vs callbacks, sync vs async, thread safety, dropping slow consumers, typing events, filtering, generics. The last few are open-ended mini-projects that culminate in a real in-memory pub/sub broker and a bridge between two brokers.

Run every solution with go vet ./..., go test ./..., and go test -race ./... before moving on. Most Observer bugs surface only under the race detector — never trust a solution you haven't raced. Each task is self-contained — copy the solution into a fresh directory, go mod init scratch, then iterate.

You need Go 1.21 or later. Tasks 9–15 use generics. Task 12 uses github.com/fsnotify/fsnotify. Task 13 uses os/signal. The remaining tasks need only the standard library.

Task 1: Basic Subject/Observer (warm-up)¶

A trivial temperature sensor that notifies registered observers when its reading changes. The subject has two operations: Subscribe(Observer) and SetTemp(float64). Each observer's Update(float64) is called synchronously on change.

sensor := &TempSensor{}
sensor.Subscribe(&Printer{Name: "A"})
sensor.Subscribe(&Printer{Name: "B"})

sensor.SetTemp(21.5)
// Output:
// A: 21.5
// B: 21.5

Acceptance criteria

• Observer interface with one method: Update(t float64).
• TempSensor holds a []Observer and a current float.
• Subscribe(o Observer) appends to the slice.
• SetTemp(t float64) updates the current value and calls Update(t) on every observer in order.
• Printer is a concrete observer that prints "<name>: <temp>".
• A main() exercises two printers receiving the same notification.

package main

import "fmt"

type Observer interface {
    Update(t float64)
}

type TempSensor struct {
    observers []Observer
    current   float64
}

func (s *TempSensor) Subscribe(o Observer) {
    s.observers = append(s.observers, o)
}

func (s *TempSensor) SetTemp(t float64) {
    s.current = t
    for _, o := range s.observers {
        o.Update(t)
    }
}

type Printer struct{ Name string }

func (p *Printer) Update(t float64) {
    fmt.Printf("%s: %.1f\n", p.Name, t)
}

func main() {
    sensor := &TempSensor{}
    sensor.Subscribe(&Printer{Name: "A"})
    sensor.Subscribe(&Printer{Name: "B"})
    sensor.SetTemp(21.5)
}

Discussion. This is the entire Observer pattern in 25 lines. Notice three properties: the subject knows observers only through the Observer interface (not their concrete types); the subject owns the list (observers don't register themselves); and notification is push, not pull — the subject hands the data to each observer rather than letting them ask.

Two things this version doesn't handle: there's no way to unsubscribe (the slice grows forever), and a panicking observer crashes the whole SetTemp call. Both are real defects in production code; tasks 2 and 4 fix them. But before you fix anything, you should be able to draw this shape from memory — it's the skeleton every later variation hangs off.

Task 2: Subscribe returns an Unsubscribe func¶

The slice from Task 1 grows forever. Fix it: change Subscribe to return an Unsubscribe function. Calling that function removes the observer from the list. Calling it twice is a no-op.

sensor := &TempSensor{}
stop := sensor.Subscribe(&Printer{Name: "A"})
sensor.Subscribe(&Printer{Name: "B"})

sensor.SetTemp(20.0) // A and B both print
stop()
sensor.SetTemp(21.0) // only B prints
stop()               // no-op, no panic

Acceptance criteria

• Subscribe(o Observer) func() returns a cancellation function.
• The returned function removes the observer from the list.
• Calling it more than once is safe (no panic, no double-remove).
• Internally, store observers in a map[uint64]Observer keyed by an auto-incremented ID — slice removal by value is fragile because Observer is an interface and equality is iffy.
• A test subscribes three observers, unsubscribes the middle one, asserts the other two still receive notifications.

package main

import (
    "fmt"
    "sort"
)

type Observer interface {
    Update(t float64)
}

type TempSensor struct {
    observers map[uint64]Observer
    nextID    uint64
}

func NewTempSensor() *TempSensor {
    return &TempSensor{observers: map[uint64]Observer{}}
}

func (s *TempSensor) Subscribe(o Observer) func() {
    s.nextID++
    id := s.nextID
    s.observers[id] = o
    return func() {
        delete(s.observers, id)
    }
}

func (s *TempSensor) SetTemp(t float64) {
    ids := make([]uint64, 0, len(s.observers))
    for id := range s.observers {
        ids = append(ids, id)
    }
    sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
    for _, id := range ids {
        s.observers[id].Update(t)
    }
}

type Printer struct{ Name string }

func (p *Printer) Update(t float64) {
    fmt.Printf("%s: %.1f\n", p.Name, t)
}

func main() {
    s := NewTempSensor()
    stop := s.Subscribe(&Printer{Name: "A"})
    s.Subscribe(&Printer{Name: "B"})

    s.SetTemp(20.0)
    stop()
    s.SetTemp(21.0)
    stop() // idempotent
}

Discussion. The "Subscribe returns Unsubscribe" idiom is what every well-designed Go pub/sub API converges on — context.WithCancel, signal.NotifyContext, prometheus.Register/Unregister, even time.AfterFunc (where the returned *Timer has Stop()). The caller holds the lifetime token. If they drop it, they've leaked a subscription; the linter can't catch that, but at least the API didn't make it ambiguous how to clean up.

delete on a missing key is a no-op in Go, which is why the second stop() call doesn't panic. Don't add an explicit _, ok := s.observers[id] check — the language already gives you idempotence for free.

Task 3: Channel-based pub/sub¶

Drop the Observer interface entirely. Each subscriber gets a <-chan float64 and reads from it directly. Subscribe creates a fresh channel, registers it internally, and returns it. Publish sends the value to every subscriber's channel.

hub := NewHub()
ch1 := hub.Subscribe()
ch2 := hub.Subscribe()

go func() { for v := range ch1 { fmt.Println("1:", v) } }()
go func() { for v := range ch2 { fmt.Println("2:", v) } }()

hub.Publish(1.0)
hub.Publish(2.0)

Acceptance criteria

• Hub holds a slice (or map) of chan float64.
• Subscribe() <-chan float64 creates a buffered channel (capacity 16) and stores it, returning the receive-only end.
• Publish(v float64) sends v to every channel. If a channel is full, block (we'll fix that in task 6).
• Close() closes every channel and clears the list.
• A test creates two subscribers, publishes 5 values, and asserts both receive all 5 in order.

package main

import (
    "fmt"
    "sync"
    "time"
)

type Hub struct {
    mu      sync.Mutex
    subs    []chan float64
    closed  bool
}

func NewHub() *Hub { return &Hub{} }

func (h *Hub) Subscribe() <-chan float64 {
    h.mu.Lock()
    defer h.mu.Unlock()
    if h.closed {
        ch := make(chan float64)
        close(ch)
        return ch
    }
    ch := make(chan float64, 16)
    h.subs = append(h.subs, ch)
    return ch
}

func (h *Hub) Publish(v float64) {
    h.mu.Lock()
    subs := make([]chan float64, len(h.subs))
    copy(subs, h.subs)
    h.mu.Unlock()
    for _, ch := range subs {
        ch <- v
    }
}

func (h *Hub) Close() {
    h.mu.Lock()
    defer h.mu.Unlock()
    if h.closed {
        return
    }
    h.closed = true
    for _, ch := range h.subs {
        close(ch)
    }
    h.subs = nil
}

func main() {
    h := NewHub()
    ch := h.Subscribe()
    go func() {
        for v := range ch {
            fmt.Println("got:", v)
        }
    }()
    h.Publish(1.0)
    h.Publish(2.0)
    time.Sleep(50 * time.Millisecond)
    h.Close()
}

Discussion. Channel-based pub/sub trades the Observer interface for a typed channel. Three benefits: subscribers compose with select, for range, and context.Done() without any extra machinery; the channel close is a natural broadcast signal ("hub is shutting down"); and the type system rejects the wrong payload at compile time.

The cost is that subscribers are now responsible for draining their channel. A slow subscriber wedges the whole Publish. Task 6 fixes this with a non-blocking send and a drop policy. A second cost is that you can't filter or transform inside the subscription cheaply — the channel is "dumb". Task 8 layers filtering back on top.

Notice the snapshot-then-iterate trick in Publish: copy the slice under the lock, release the lock, then send. Holding the lock across ch <- v would let one slow subscriber block all subscriptions and unsubscriptions. This pattern repeats in tasks 4 and 14.

Task 4: Thread-safe observer list¶

Return to Task 2's struct-with-map shape, but now make it safe under concurrent Subscribe, Unsubscribe, and Notify. The race detector must be silent.

s := NewSafeSensor()
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        stop := s.Subscribe(&Printer{})
        s.SetTemp(float64(i))
        stop()
    }()
}
wg.Wait()

Acceptance criteria

• SafeSensor has a sync.RWMutex.
• Subscribe takes the write lock; so does the returned unsubscribe closure.
• SetTemp (the notify path) takes the read lock briefly to snapshot the observer list, then releases it before calling Update. Never hold a lock across user code.
• go test -race ./... passes on 100 concurrent subscribe/notify/unsubscribe calls.
• A test asserts the race detector is clean under stress.

package main

import (
    "sync"
    "sync/atomic"
)

type Observer interface {
    Update(t float64)
}

type SafeSensor struct {
    mu        sync.RWMutex
    observers map[uint64]Observer
    nextID    uint64
}

func NewSafeSensor() *SafeSensor {
    return &SafeSensor{observers: map[uint64]Observer{}}
}

func (s *SafeSensor) Subscribe(o Observer) func() {
    id := atomic.AddUint64(&s.nextID, 1)
    s.mu.Lock()
    s.observers[id] = o
    s.mu.Unlock()
    return func() {
        s.mu.Lock()
        delete(s.observers, id)
        s.mu.Unlock()
    }
}

func (s *SafeSensor) SetTemp(t float64) {
    s.mu.RLock()
    snapshot := make([]Observer, 0, len(s.observers))
    for _, o := range s.observers {
        snapshot = append(snapshot, o)
    }
    s.mu.RUnlock()
    for _, o := range snapshot {
        o.Update(t)
    }
}

Discussion. Two rules to internalize. First: never hold a lock across user-provided code. The observer's Update may take seconds, may panic, may call back into the subject. None of these are crimes the subject should commit on the user's behalf. Snapshot under the lock; iterate without it.

Second: atomic.AddUint64 for the ID counter, not the lock. You could put the increment inside the lock, but separating it makes the critical section smaller and the intent obvious — "give me a unique ID" is conceptually independent of "register this observer".

A subtle hazard the test won't catch: between the snapshot and the iteration, an unsubscribed observer can still receive one final Update. That's fine for most APIs (it's a "best effort" delivery), but if the observer's Update accesses freed state, you have a use-after-free. The defensive answer is for observers to guard their own internal state — Task 5 makes this explicit by switching to async delivery, which has the same race built in.

Task 5: Async notification (one goroutine per observer)¶

In Task 4, a slow observer blocks every other observer because SetTemp calls them in a loop. Change SetTemp to fire each Update in its own goroutine. The notify call returns immediately; observers run concurrently.

s := NewAsyncSensor()
s.Subscribe(&SleepyObserver{Delay: 100 * time.Millisecond})
s.Subscribe(&SleepyObserver{Delay: 100 * time.Millisecond})

start := time.Now()
s.SetTemp(20.0)
fmt.Println("returned in", time.Since(start)) // ~0ms, not 200ms

Acceptance criteria

• SetTemp launches one goroutine per observer and returns immediately.
• Provide a Wait() method that blocks until all in-flight notifications complete. (Use a sync.WaitGroup you increment in SetTemp and decrement at the end of each goroutine.)
• Panics in an observer must not crash the program. Recover inside each goroutine and log the panic.
• A test with two 100ms observers asserts SetTemp returns in under 10ms but Wait blocks for ~100ms.

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

type Observer interface {
    Update(t float64)
}

type AsyncSensor struct {
    mu        sync.RWMutex
    observers map[uint64]Observer
    nextID    uint64
    wg        sync.WaitGroup
}

func NewAsyncSensor() *AsyncSensor {
    return &AsyncSensor{observers: map[uint64]Observer{}}
}

func (s *AsyncSensor) Subscribe(o Observer) func() {
    id := atomic.AddUint64(&s.nextID, 1)
    s.mu.Lock()
    s.observers[id] = o
    s.mu.Unlock()
    return func() {
        s.mu.Lock()
        delete(s.observers, id)
        s.mu.Unlock()
    }
}

func (s *AsyncSensor) SetTemp(t float64) {
    s.mu.RLock()
    snapshot := make([]Observer, 0, len(s.observers))
    for _, o := range s.observers {
        snapshot = append(snapshot, o)
    }
    s.mu.RUnlock()

    for _, o := range snapshot {
        s.wg.Add(1)
        go func(o Observer) {
            defer s.wg.Done()
            defer func() {
                if r := recover(); r != nil {
                    fmt.Println("observer panicked:", r)
                }
            }()
            o.Update(t)
        }(o)
    }
}

func (s *AsyncSensor) Wait() { s.wg.Wait() }

Discussion. Goroutine-per-notification is the cheapest concurrency model and the most dangerous. Cheap because goroutines cost ~2KB and the scheduler handles backpressure-free. Dangerous because a flood of SetTemp calls during a slow observer fans out into thousands of stacked goroutines, each blocked on the same observer. The observer becomes the bottleneck and the memory leak.

There are three escape routes from this trap, in increasing order of sophistication: drop notifications to slow observers (Task 6), bound the work queue per observer (Task 10), or serialize notifications per observer through a channel they own (Task 11). All three accept that a slow observer is permanently slow and that the system must degrade gracefully rather than amplify the slowness.

The defer recover() inside each goroutine is non-negotiable. Without it, a single bad observer kills the program — goroutine panics propagate to the runtime, not the spawning code. If you've ever seen a Go service crash with "goroutine X panicked" and no obvious caller, this is the shape.

Task 6: Slow observer dropping¶

In Task 3's channel-based hub, a slow subscriber wedges every Publish. Fix it: change Publish to use a non-blocking send (select with default). If the channel is full, drop the value and increment a per-subscriber drop counter.

hub := NewDroppingHub(4) // capacity 4 per subscriber
ch := hub.Subscribe()

for i := 0; i < 100; i++ {
    hub.Publish(i) // most will drop because ch isn't being drained
}
fmt.Println("drops:", hub.Drops())

Acceptance criteria

• NewDroppingHub(capacity int) sets the per-subscriber buffer size.
• Publish uses select { case ch <- v: default: } — never blocks.
• Per-subscriber drop count is exposed (a map[uint64]uint64 or a counter on the subscriber).
• A test publishes 1000 values to an undrained channel of capacity 16 and asserts drops > 900.

package main

import (
    "sync"
    "sync/atomic"
)

type subscription struct {
    ch    chan int
    drops atomic.Uint64
}

type DroppingHub struct {
    mu       sync.RWMutex
    subs     map[uint64]*subscription
    nextID   uint64
    capacity int
}

func NewDroppingHub(capacity int) *DroppingHub {
    return &DroppingHub{
        subs:     map[uint64]*subscription{},
        capacity: capacity,
    }
}

func (h *DroppingHub) Subscribe() (<-chan int, func()) {
    id := atomic.AddUint64(&h.nextID, 1)
    sub := &subscription{ch: make(chan int, h.capacity)}
    h.mu.Lock()
    h.subs[id] = sub
    h.mu.Unlock()
    return sub.ch, func() {
        h.mu.Lock()
        if s, ok := h.subs[id]; ok {
            close(s.ch)
            delete(h.subs, id)
        }
        h.mu.Unlock()
    }
}

func (h *DroppingHub) Publish(v int) {
    h.mu.RLock()
    subs := make([]*subscription, 0, len(h.subs))
    for _, s := range h.subs {
        subs = append(subs, s)
    }
    h.mu.RUnlock()
    for _, s := range subs {
        select {
        case s.ch <- v:
        default:
            s.drops.Add(1)
        }
    }
}

func (h *DroppingHub) Drops() uint64 {
    h.mu.RLock()
    defer h.mu.RUnlock()
    var total uint64
    for _, s := range h.subs {
        total += s.drops.Load()
    }
    return total
}

Discussion. Non-blocking sends are how Go pub/sub stays alive under load. The classic mistake — for _, ch := range subs { ch <- v } — couples every publisher's latency to the slowest subscriber's latency. One stuck subscriber, and the whole bus halts. The select { case ch <- v: default: } idiom transfers responsibility back to the subscriber: keep up, or lose data. It's an explicit contract, and it's the right default for telemetry, metrics, and any "best effort" stream.

Two policies you'll meet in the wild. Drop newest (what we did): the most recent value is dropped. Simplest, but loses freshness — the subscriber's view of the world goes stale. Drop oldest (head-of-queue): you <-ch then send, evicting the oldest. Keeps freshness but requires a slightly bigger lock window. Pick based on whether the data is incremental (drop newest is fine — counters, deltas) or absolute (drop oldest — last known position, current price).

The drop counter is not optional. A pub/sub system that silently drops messages is a debugging black hole. Exposing the count gives ops a knob: alert if drops > N/sec, or auto-resize the buffer.

Task 7: Typed events (multiple event types)¶

A single subject publishes multiple event types: UserCreated, UserDeleted, OrderPlaced. Subscribers register for a specific event type and only receive that type.

bus := NewTypedBus()
bus.On("UserCreated", func(e Event) { fmt.Println("create:", e) })
bus.On("OrderPlaced", func(e Event) { fmt.Println("order:", e) })

bus.Emit("UserCreated", Event{"id": 42})
bus.Emit("OrderPlaced", Event{"sku": "X"})
bus.Emit("UnknownType", Event{}) // no-op, no error

Acceptance criteria

• Event is a map[string]any.
• Handler is func(Event).
• TypedBus.On(eventType string, h Handler) func() returns an unsubscribe function.
• TypedBus.Emit(eventType string, e Event) calls every handler registered for that type, in registration order.
• Emit on an unknown type is a no-op (no panic, no error).
• A test asserts that subscribing to "A" then emitting "B" produces no calls; emitting "A" calls only the "A" subscribers.

package main

import (
    "sort"
    "sync"
    "sync/atomic"
)

type Event map[string]any
type Handler func(Event)

type TypedBus struct {
    mu       sync.RWMutex
    handlers map[string]map[uint64]Handler
    nextID   uint64
}

func NewTypedBus() *TypedBus {
    return &TypedBus{handlers: map[string]map[uint64]Handler{}}
}

func (b *TypedBus) On(eventType string, h Handler) func() {
    id := atomic.AddUint64(&b.nextID, 1)
    b.mu.Lock()
    if b.handlers[eventType] == nil {
        b.handlers[eventType] = map[uint64]Handler{}
    }
    b.handlers[eventType][id] = h
    b.mu.Unlock()
    return func() {
        b.mu.Lock()
        defer b.mu.Unlock()
        bucket := b.handlers[eventType]
        if bucket == nil {
            return
        }
        delete(bucket, id)
        if len(bucket) == 0 {
            delete(b.handlers, eventType)
        }
    }
}

func (b *TypedBus) Emit(eventType string, e Event) {
    b.mu.RLock()
    bucket := b.handlers[eventType]
    if bucket == nil {
        b.mu.RUnlock()
        return
    }
    ids := make([]uint64, 0, len(bucket))
    for id := range bucket {
        ids = append(ids, id)
    }
    sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
    snapshot := make([]Handler, len(ids))
    for i, id := range ids {
        snapshot[i] = bucket[id]
    }
    b.mu.RUnlock()
    for _, h := range snapshot {
        h(e)
    }
}

Discussion. The string-keyed event bus is the most popular Observer variant outside Go — it's how EventEmitter in Node.js, signals in Django, and the DOM event system all work. The string key is dynamic, which is good for plugin systems and bad for type safety: typo "UserCreted" and your handler silently never fires. Tests catch most of these; some hide for months.

In Go we have a better tool — generics — and Task 9 will use it. But the string-keyed form is still right for systems where the event types are open (plugins, scripted hooks, message bus over a network where the type comes from the wire). The lesson: pick the most static representation that still fits your extension model.

One bug magnet: when you call Emit for an unknown type, returning silently is correct but invisible. If you suspect a typo, add a debug logger that prints unmatched emits in dev builds. Mature event buses have an "unhandled" hook for exactly this.

Task 8: Filtered subscriptions (predicate)¶

Layer a filter on top of Task 7's typed bus. Subscribers can attach a predicate func(Event) bool and only receive events that pass.

bus := NewFilteredBus()
bus.OnIf("OrderPlaced", func(e Event) bool { return e["amount"].(int) > 100 },
    func(e Event) { fmt.Println("big order:", e) })

bus.Emit("OrderPlaced", Event{"amount": 50})  // skipped
bus.Emit("OrderPlaced", Event{"amount": 500}) // matched

Acceptance criteria

• OnIf(eventType string, pred func(Event) bool, h Handler) func() adds a filtered subscription.
• The plain On is equivalent to OnIf with a predicate that always returns true.
• Predicates run inside the broker, on the publisher's goroutine — they must be cheap and side-effect-free.
• A panicking predicate is recovered (treat as "no match") so one bad subscriber can't crash the whole emit.
• A test asserts predicates are called once per emit-per-subscriber, regardless of how many handlers exist.

package main

import (
    "fmt"
    "sort"
    "sync"
    "sync/atomic"
)

type Event map[string]any
type Handler func(Event)
type Predicate func(Event) bool

type sub struct {
    pred Predicate
    h    Handler
}

type FilteredBus struct {
    mu       sync.RWMutex
    subs     map[string]map[uint64]sub
    nextID   uint64
}

func NewFilteredBus() *FilteredBus {
    return &FilteredBus{subs: map[string]map[uint64]sub{}}
}

var alwaysTrue Predicate = func(Event) bool { return true }

func (b *FilteredBus) On(eventType string, h Handler) func() {
    return b.OnIf(eventType, alwaysTrue, h)
}

func (b *FilteredBus) OnIf(eventType string, pred Predicate, h Handler) func() {
    id := atomic.AddUint64(&b.nextID, 1)
    b.mu.Lock()
    if b.subs[eventType] == nil {
        b.subs[eventType] = map[uint64]sub{}
    }
    b.subs[eventType][id] = sub{pred: pred, h: h}
    b.mu.Unlock()
    return func() {
        b.mu.Lock()
        defer b.mu.Unlock()
        bucket := b.subs[eventType]
        if bucket == nil {
            return
        }
        delete(bucket, id)
        if len(bucket) == 0 {
            delete(b.subs, eventType)
        }
    }
}

func matches(pred Predicate, e Event) (ok bool) {
    defer func() {
        if r := recover(); r != nil {
            fmt.Println("predicate panic, skipping:", r)
            ok = false
        }
    }()
    return pred(e)
}

func (b *FilteredBus) Emit(eventType string, e Event) {
    b.mu.RLock()
    bucket := b.subs[eventType]
    if bucket == nil {
        b.mu.RUnlock()
        return
    }
    ids := make([]uint64, 0, len(bucket))
    for id := range bucket {
        ids = append(ids, id)
    }
    sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
    snapshot := make([]sub, len(ids))
    for i, id := range ids {
        snapshot[i] = bucket[id]
    }
    b.mu.RUnlock()
    for _, s := range snapshot {
        if matches(s.pred, e) {
            s.h(e)
        }
    }
}

Discussion. Filtering at the broker is a real lever. The alternative — filter inside the handler — produces correct behavior but wastes work: every "OrderPlaced" handler runs even when 99% of orders are small. With predicates, the broker can short-circuit before invoking the handler, and the math gets dramatic when handlers are expensive (DB writes, HTTP calls).

The trade-off is where the cost lives. Predicates run on the publisher's goroutine, blocking the emit. For 1000 subscribers with cheap predicates this is fine. For one subscriber with a 10ms predicate (regex match, JSON path query), it's a disaster — the publisher's loop is now bottlenecked on a consumer concern. The rule: predicates must be O(1) field comparisons. Anything else belongs in the handler.

A more advanced design pushes the filter expression as data, not code — RxJava's filter, NATS's subject hierarchy (orders.>, orders.us.*), Kafka's topic + partition key. The broker can then index subscriptions by filter and dispatch in O(matching subs) instead of O(all subs). We won't go that far here; Task 14 takes the first step.

Task 9: Generic Observer[T]¶

Replace the string keys from Task 7 with Go generics. Each event type is a Go type. Publishing a UserCreated value routes to handlers of Observer[UserCreated].

type UserCreated struct{ ID int }
type OrderPlaced struct{ SKU string }

bus := NewGenericBus()
Subscribe(bus, func(e UserCreated) { fmt.Println("user:", e.ID) })
Subscribe(bus, func(e OrderPlaced) { fmt.Println("order:", e.SKU) })

Publish(bus, UserCreated{ID: 42})
Publish(bus, OrderPlaced{SKU: "X"})

Acceptance criteria

• GenericBus is a struct holding map[reflect.Type]any, where any is []func(T) for the appropriate T.
• Subscribe[T](bus, h func(T)) func() registers a typed handler.
• Publish[T](bus, e T) looks up the bucket for T and calls every handler.
• Compile-time error if you Publish[Foo] but only Subscribe[Bar]. (Actually: it's a runtime no-op, since there are no subscribers. The compile-time check is on the handler signature, not the cross-product.)
• A test confirms Subscribe[UserCreated] never receives an OrderPlaced.

package main

import (
    "reflect"
    "sync"
)

type GenericBus struct {
    mu       sync.RWMutex
    handlers map[reflect.Type]any
}

func NewGenericBus() *GenericBus {
    return &GenericBus{handlers: map[reflect.Type]any{}}
}

func typeKey[T any]() reflect.Type {
    return reflect.TypeOf((*T)(nil)).Elem()
}

func Subscribe[T any](b *GenericBus, h func(T)) func() {
    k := typeKey[T]()
    b.mu.Lock()
    hs, _ := b.handlers[k].([]func(T))
    hs = append(hs, h)
    b.handlers[k] = hs
    idx := len(hs) - 1
    b.mu.Unlock()
    return func() {
        b.mu.Lock()
        defer b.mu.Unlock()
        hs, _ := b.handlers[k].([]func(T))
        if idx >= len(hs) {
            return
        }
        // Tombstone: replace with nil instead of slice-tricks to keep IDs stable.
        hs[idx] = nil
        b.handlers[k] = hs
    }
}

func Publish[T any](b *GenericBus, e T) {
    k := typeKey[T]()
    b.mu.RLock()
    hs, _ := b.handlers[k].([]func(T))
    snapshot := make([]func(T), len(hs))
    copy(snapshot, hs)
    b.mu.RUnlock()
    for _, h := range snapshot {
        if h != nil {
            h(e)
        }
    }
}

Discussion. Go generics give Observer the type safety it was missing. Publish[UserCreated] only routes to Subscribe[UserCreated]; the compiler enforces signature shape; refactoring a field on UserCreated updates every handler instead of silently mismatching a string key. This is what every modern Go pub/sub library now reaches for — nats.io, watermill, even small in-process buses like dispatcher.go.

There's a real tension you should be aware of: generics force closed event types. Adding a new event type requires Go code changes, recompilation, redeployment. The string-keyed bus from Task 7 lets you bolt on a new event type by editing a config file. Both are right for different problems. For application code, generics win — typo-proof, refactor-safe, IDE-completed. For plugin systems and cross-language buses, strings win — late-bound, open-ended, language-agnostic.

The tombstone trick (nil out an unsubscribed handler instead of removing it) keeps slice indices stable across unsubscribes during emit. The cost is the slice never shrinks. For long-lived buses with churn, periodically compact (rebuild the slice without nils) when nil ratio exceeds 50%. Task 14 implements that.

Task 10: Backpressure with bounded buffer¶

Combine Task 5's async fan-out with Task 6's drop policy in a more useful shape: each subscriber owns a buffered channel of bounded capacity, drained by a per-subscriber goroutine. The broker pushes to the channel with a non-blocking send.

broker := NewBoundedBroker(16) // 16-slot buffer per subscriber
broker.Subscribe(func(e Event) {
    time.Sleep(50 * time.Millisecond) // slow!
    fmt.Println("got:", e)
})

for i := 0; i < 1000; i++ {
    broker.Publish(Event{"i": i}) // most dropped, broker stays fast
}

Acceptance criteria

• Each subscriber gets a chan Event of fixed capacity.
• A dedicated goroutine drains that channel and invokes the handler.
• Publish does a non-blocking send to each subscriber's channel.
• On full channel, increment a drop counter (one per subscriber).
• Unsubscribe closes the channel and the drain goroutine exits cleanly.
• A test publishes 1000 events to a slow subscriber and asserts drops + delivered == 1000.

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

type Event map[string]any
type Handler func(Event)

type boundedSub struct {
    ch    chan Event
    drops atomic.Uint64
}

type BoundedBroker struct {
    mu       sync.RWMutex
    subs     map[uint64]*boundedSub
    nextID   uint64
    capacity int
}

func NewBoundedBroker(capacity int) *BoundedBroker {
    return &BoundedBroker{
        subs:     map[uint64]*boundedSub{},
        capacity: capacity,
    }
}

func (b *BoundedBroker) Subscribe(h Handler) func() {
    id := atomic.AddUint64(&b.nextID, 1)
    sub := &boundedSub{ch: make(chan Event, b.capacity)}
    b.mu.Lock()
    b.subs[id] = sub
    b.mu.Unlock()
    go func() {
        defer func() {
            if r := recover(); r != nil {
                fmt.Println("handler panic:", r)
            }
        }()
        for e := range sub.ch {
            func() {
                defer func() { _ = recover() }()
                h(e)
            }()
        }
    }()
    return func() {
        b.mu.Lock()
        s, ok := b.subs[id]
        if ok {
            delete(b.subs, id)
        }
        b.mu.Unlock()
        if ok {
            close(s.ch)
        }
    }
}

func (b *BoundedBroker) Publish(e Event) {
    b.mu.RLock()
    subs := make([]*boundedSub, 0, len(b.subs))
    for _, s := range b.subs {
        subs = append(subs, s)
    }
    b.mu.RUnlock()
    for _, s := range subs {
        select {
        case s.ch <- e:
        default:
            s.drops.Add(1)
        }
    }
}

func (b *BoundedBroker) Drops() uint64 {
    b.mu.RLock()
    defer b.mu.RUnlock()
    var n uint64
    for _, s := range b.subs {
        n += s.drops.Load()
    }
    return n
}

Discussion. This is the shape used by Prometheus's scrape pool, Kubernetes's informers, and most real-world Go pub/sub buses. Per-subscriber buffer + drain goroutine + non-blocking publish. The broker is decoupled from handler latency entirely — its only job is to put events into channels and count drops.

The capacity choice is a real tuning knob. Too small: every burst triggers drops, the system feels lossy even with cooperative consumers. Too large: a slow consumer hides for a long time before drops appear, and you've buffered megabytes of stale events. A practical rule of thumb: buffer ~1 second of expected burst rate, and alert on persistent drops. If you can't size the buffer because traffic is bimodal (idle then huge spike), you probably want a different pattern — maybe a batched flush or an external queue like Kafka.

Notice how the broker's Publish time is now O(subscribers), independent of how slow any handler is. Compare to Task 5's Wait() — there, the broker's overall throughput was capped by the slowest handler because goroutines piled up. Here, the slow handler just drops events. That's a much friendlier failure mode.

Task 11: Order-preserving fan-out¶

Task 10 delivers events out of order across subscribers (because each subscriber has its own goroutine that schedules independently). For some use cases — say, replaying a state machine — every subscriber must see events in the same order as they were published, with no gaps. Build a broker that guarantees per-subscriber ordering and refuses to drop events. It blocks Publish if any subscriber's buffer is full.

broker := NewOrderedBroker(16)
ch1 := broker.Subscribe()
ch2 := broker.Subscribe()

for i := 0; i < 100; i++ {
    broker.Publish(Event{"i": i}) // blocks if any consumer is slow
}

Acceptance criteria

• Each subscriber gets a buffered channel.
• Publish sends to every channel synchronously (blocking on each).
• Each subscriber sees events in the same order as Publish calls.
• Adding a slow subscriber blocks the broker — this is documented behavior, not a bug.
• A test publishes 100 events to two subscribers and asserts both receive [0, 1, 2, ..., 99] in order.

package main

import (
    "sync"
    "sync/atomic"
)

type Event map[string]any

type OrderedBroker struct {
    inbound chan Event

    mu       sync.RWMutex
    subs     map[uint64]chan Event
    nextID   uint64
    capacity int

    closeOnce sync.Once
    done      chan struct{}
}

func NewOrderedBroker(capacity int) *OrderedBroker {
    b := &OrderedBroker{
        inbound:  make(chan Event, 64),
        subs:     map[uint64]chan Event{},
        capacity: capacity,
        done:     make(chan struct{}),
    }
    go b.dispatch()
    return b
}

func (b *OrderedBroker) Subscribe() (<-chan Event, func()) {
    id := atomic.AddUint64(&b.nextID, 1)
    ch := make(chan Event, b.capacity)
    b.mu.Lock()
    b.subs[id] = ch
    b.mu.Unlock()
    return ch, func() {
        b.mu.Lock()
        if c, ok := b.subs[id]; ok {
            close(c)
            delete(b.subs, id)
        }
        b.mu.Unlock()
    }
}

func (b *OrderedBroker) Publish(e Event) {
    select {
    case b.inbound <- e:
    case <-b.done:
    }
}

func (b *OrderedBroker) dispatch() {
    for {
        select {
        case e := <-b.inbound:
            b.mu.RLock()
            ids := make([]uint64, 0, len(b.subs))
            for id := range b.subs {
                ids = append(ids, id)
            }
            b.mu.RUnlock()
            for _, id := range ids {
                b.mu.RLock()
                ch, ok := b.subs[id]
                b.mu.RUnlock()
                if !ok {
                    continue
                }
                ch <- e // blocking — preserves ordering
            }
        case <-b.done:
            return
        }
    }
}

func (b *OrderedBroker) Close() {
    b.closeOnce.Do(func() {
        close(b.done)
        b.mu.Lock()
        for _, ch := range b.subs {
            close(ch)
        }
        b.subs = nil
        b.mu.Unlock()
    })
}

Discussion. Ordered delivery is the most expensive guarantee a pub/sub system can offer. Three properties co-evolve: order, no drops, and back-pressure. You can have any two, but not all three with unlimited scale. This task picks order + no drops, and accepts back-pressure: the broker's publish can block, and a slow subscriber drags the system down. That's a deliberate trade-off.

The single dispatch goroutine is the secret to per-subscriber ordering across multiple publishers. Without it, two Publish calls racing can interleave at each subscriber differently — subscriber A sees [1, 2], subscriber B sees [2, 1]. Funneling everything through inbound linearizes the order. The cost is the inbound channel becomes the new bottleneck — total throughput is capped at "one event per dispatch loop iteration".

When you outgrow this design, you partition: split the broker into N independent ordered brokers, each handling a subset of keys (hash(key) % N). Each broker preserves order within its partition; cross-partition order is sacrificed for parallelism. Kafka's partitions are exactly this. NATS's JetStream is similar. The trade-off appears in your application: pick a partition key that groups events that must stay ordered (per user, per account) and let other events flow concurrently.

Task 12: Watching file changes (fsnotify-style)¶

Build a small watcher that observes a directory for new files and notifies subscribers. Use github.com/fsnotify/fsnotify for the OS-level events; expose a clean Observer API on top.

w, _ := NewFileWatcher("./logs")
defer w.Close()

w.OnCreate(func(path string) { fmt.Println("created:", path) })
w.OnDelete(func(path string) { fmt.Println("deleted:", path) })

w.Start() // blocks until Close