package main import ( "fmt" "sync" "time" ) // Broadcaster manages multiple listener channels type Broadcaster struct { listeners []chan interface{} mu sync.Mutex } // NewBroadcaster creates a new Broadcaster instance func NewBroadcaster() *Broadcaster { return &Broadcaster{ listeners: make([]chan interface{}, 0), } } // Register adds a new listener channel func (b *Broadcaster) Register(listener chan interface{}) { b.mu.Lock() defer b.mu.Unlock() b.listeners = append(b.listeners, listener) } // Deregister removes a listener channel func (b *Broadcaster) Deregister(listener chan interface{}) { b.mu.Lock() defer b.mu.Unlock() for i, l := range b.listeners { if l == listener { b.listeners = append(b.listeners[:i], b.listeners[i+1:]...) close(l) // It's good practice to close the channel when deregistering return } } } // Broadcast sends a message to all registered listeners func (b *Broadcaster) Broadcast(message interface{}) { b.mu.Lock() defer b.mu.Unlock() for _, listener := range b.listeners { select { case listener <- message: default: // Non-blocking send, drop message if listener is not ready fmt.Printf("Warning: Listener not ready for message: %v ", message) } } } func main() { b := NewBroadcaster() // Create listener channels listener1 := make(chan interface{}, 1) // Buffered for non-blocking broadcast listener2 := make(chan interface{}, 1) listener3 := make(chan interface{}, 0) // Unbuffered to demonstrate default case b.Register(listener1) b.Register(listener2) b.Register(listener3) // This listener might drop messages due to unbuffered and immediate send var wg sync.WaitGroup // Start goroutines to listen for messages wg.Add(3) go func(id int, l <-chan interface{}) { defer wg.Done() for msg := range l { fmt.Printf("Listener %d received: %v ", id, msg) } fmt.Printf("Listener %d finished. ", id) }(1, listener1) go func(id int, l <-chan interface{}) { defer wg.Done() for msg := range l { fmt.Printf("Listener %d received: %v ", id, msg) } fmt.Printf("Listener %d finished. ", id) }(2, listener2) go func(id int, l <-chan interface{}) { // This listener will only receive if it's immediately ready defer wg.Done() for msg := range l { fmt.Printf("Listener %d received: %v ", id, msg) } fmt.Printf("Listener %d finished. ", id) }(3, listener3) // Broadcast some messages time.Sleep(10 * time.Millisecond) // Give listeners time to start b.Broadcast("Event A") b.Broadcast(123) b.Broadcast(true) time.Sleep(50 * time.Millisecond) // Give time for messages to be processed b.Deregister(listener1) // Remove listener1 fmt.Println("Deregistered listener 1.") b.Broadcast("Event B - after deregister") b.Deregister(listener2) b.Deregister(listener3) // Also deregister the last listener // All listeners are deregistered and closed, so goroutines should exit wg.Wait() fmt.Println("All listeners finished, main exiting.") }