Pub/Sub — Find the Bug¶

type Broker struct {
    mu   sync.Mutex
    subs map[string][]chan Event
}

func (b *Broker) Publish(topic string, e Event) {
    b.mu.Lock()
    defer b.mu.Unlock()
    for _, ch := range b.subs[topic] {
        ch <- e                              // blocks while holding the lock
    }
}

func (b *Broker) Subscribe(topic string) <-chan Event {
    ch := make(chan Event)                   // UNBUFFERED
    b.mu.Lock()
    b.subs[topic] = append(b.subs[topic], ch)
    b.mu.Unlock()
    return ch
}

func (b *Broker) Publish(topic string, e Event) {
    // snapshot subscribers as cps...
    for _, ch := range cps {
        ch <- e                              // blocks until receiver ready
    }
}

func handler(w http.ResponseWriter, r *http.Request) {
    ch := broker.Subscribe("notifications")
    for ev := range ch {
        fmt.Fprintf(w, "data: %s\n\n", ev.JSON)
        if f, ok := w.(http.Flusher); ok { f.Flush() }
    }
}

func (b *Broker) Close() {
    b.mu.Lock()
    defer b.mu.Unlock()
    b.closed = true
    b.subs = nil                             // drop refs; channels never closed
}

// subscriber elsewhere: for ev := range ch { process(ev) }

func placeOrder(o *Order) {
    ev := &OrderEvent{OrderID: o.ID, Status: "placed", Items: o.Items}
    broker.Publish("orders", ev)

    if o.NeedsFraudCheck() {
        ev.Status = "pending_review"         // mutate after publish
    }
}
// audit subscriber:   log.Printf("order %s status=%s", ev.OrderID, ev.Status)
// webhook subscriber: sendWebhook(ev)

type Broker struct {
    subs map[string][]chan Event              // no mutex
}

func (b *Broker) Subscribe(topic string) <-chan Event {
    ch := make(chan Event, 32)
    b.subs[topic] = append(b.subs[topic], ch)
    return ch
}

func (b *Broker) Publish(topic string, e Event) {
    for _, ch := range b.subs[topic] {
        select { case ch <- e: default: }
    }
}

func (b *Broker) Unsubscribe(topic string, target <-chan Event) {
    b.mu.Lock()
    defer b.mu.Unlock()
    list := b.subs[topic]
    for i, ch := range list {
        if ch == target {
            b.subs[topic] = append(list[:i], list[i+1:]...)
            return                           // never closes the channel
        }
    }
}

// worker: for ev := range ch { process(ev) }   // never exits

// "orders.*" matches "orders.placed"; "*" matches everything
func match(pattern, topic string) bool {
    if pattern == "*" { return true }
    if strings.HasSuffix(pattern, ".*") {
        prefix := pattern[:len(pattern)-2]   // strips ".*"
        return strings.HasPrefix(topic, prefix)
    }
    return pattern == topic
}

func consume(sub *nats.Subscription) {
    for {
        msg, err := sub.NextMsg(time.Second)
        if err != nil { continue }

        if err := msg.Ack(); err != nil {
            log.Print(err); continue
        }
        if err := process(msg.Data); err != nil {
            log.Printf("process: %v", err)
        }
    }
}

// go.mod: go 1.21
for _, s := range cps {
    go func() {
        select {
        case s.ch <- e:                      // s captured by reference
        case <-time.After(s.deadline):
        }
    }()
}

func (c *NATSClient) onDisconnect() {
    c.mu.Lock()
    defer c.mu.Unlock()
    for {
        conn, err := nats.Connect(c.url, nats.Timeout(30*time.Second))
        if err == nil { c.conn = conn; return }
        time.Sleep(time.Second)
    }
}

func (c *NATSClient) Publish(topic string, data []byte) error {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.conn.Publish(topic, data)
}

return kafka.NewReader(kafka.ReaderConfig{
    Brokers:       brokers,
    GroupID:       group,
    Topic:         topic,
    QueueCapacity: 1000,                     // prefetch a lot
    MinBytes:      10 << 10,
    MaxBytes:      10 << 20,
})

// handler averages 200ms; occasionally 5s under load

func (c *Consumer) Run(ctx context.Context) error {
    for {
        msg, err := c.reader.FetchMessage(ctx)   // manual commit mode
        if err != nil { return err }

        if err := c.processInTx(msg); err != nil {
            log.Printf("process: %v", err); continue
        }
        // BUG: c.reader.CommitMessages(ctx, msg) never called
    }
}

func (b *Broker) Publish(topic string, e Event) {
    // snapshot subscribers as cps...
    for _, fn := range cps {
        fn(e)                                // synchronous, in the publisher's goroutine
    }
}

func startBus() {
    go func() {
        for e := range incoming { broker.Publish(e.Topic, e) }
    }()
}

type CloudEvent struct {
    ID, Source, Type string
    Data             map[string]any
}

func publishUserAction(action, userID string, payload map[string]any) error {
    ev := CloudEvent{
        ID:     uuid.NewString(),
        Source: "/users",
        Type:   "com.acme.user." + action,   // action from HTTP query
        Data:   payload,
    }
    return nats.Publish("audit", ev)
}

func handleAction(w http.ResponseWriter, r *http.Request) {
    action := r.URL.Query().Get("action")    // attacker-controlled
    publishUserAction(action, userID, body)
}