Broadcast Events to Slow and Fast Subscribers

package main

import (
	"fmt"
	"sync"
)

type Broker struct {
	mu          sync.RWMutex
	subscribers map[int]chan string
	nextID      int
}

func NewBroker() *Broker {
	return &Broker{subscribers: make(map[int]chan string)}
}

func (b *Broker) Subscribe(buffer int) (int, <-chan string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	id := b.nextID
	b.nextID++
	channel := make(chan string, buffer)
	b.subscribers[id] = channel
	return id, channel
}

func (b *Broker) Unsubscribe(id int) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if channel, ok := b.subscribers[id]; ok {
		delete(b.subscribers, id)
		close(channel)
	}
}

func (b *Broker) Publish(event string) int {
	b.mu.RLock()
	defer b.mu.RUnlock()
	delivered := 0
	for _, channel := range b.subscribers {
		select {
		case channel <- event:
			delivered++
		default:
		}
	}
	return delivered
}

func main() {
	broker := NewBroker()
	fastID, fast := broker.Subscribe(4)
	defer broker.Unsubscribe(fastID)
	slowID, slow := broker.Subscribe(1)
	defer broker.Unsubscribe(slowID)

	delivered := []int{broker.Publish("a"), broker.Publish("b"), broker.Publish("c")}
	fmt.Println("delivered:", delivered)
	if delivered[0] != 2 || delivered[1] != 1 || delivered[2] != 1 {
		panic("unexpected delivery counts")
	}

	fastSeen := []string{<-fast, <-fast, <-fast}
	slowSeen := []string{<-slow}
	fmt.Println(fastSeen, slowSeen)
	if slowSeen[0] != "a" {
		panic("slow subscriber should keep only the first event")
	}

	broker.Unsubscribe(slowID)
	if broker.Publish("d") != 1 {
		panic("expected one subscriber after unsubscribe")
	}
	broker.Unsubscribe(fastID)
	if broker.Publish("e") != 0 {
		panic("expected no subscribers")
	}
}