Non-blocking fan-out broker in go

broker.go

package main

import (
	"fmt"
	"time"
)

type Message struct {
	Timestamp time.Time
	Client *Client
	Value string
}

type Client struct {
	Inbox <-chan []*Message
	Outbox chan<- *Message
}

type Broker struct {
	Timeout time.Duration
	reqadd chan chan<- *Client
	reqremove chan *Client
}

func NewBroker() *Broker {
	b := &Broker{
		Timeout: 60 * time.Second,
		reqadd: make(chan chan<- *Client),
		reqremove: make(chan *Client),
	}
	go func() {
		var outbox = make(chan *Message, 64)
		var clients = make(map[*Client]chan []*Message)
		for {
			select {
			case reply := <-b.reqadd:
				inbox := make(chan []*Message, 1)
				client := &Client{
					Inbox: inbox,
					Outbox: outbox,
				}
				clients[client] = inbox
				reply <- client
			case client := <-b.reqremove:
				if inbox, ok := clients[client]; ok {
					delete(clients, client)
					close(inbox)
				}
			case message := <-outbox:
				// Slow clients scheduled to be discarded
				var slow []*Client
				// Inbox older than this is deemed slow
				expired := time.Now().Add(-b.Timeout)
				// Iterate over all clients and their inboxes
				for client, inbox := range(clients) {
					if client == message.Client {
						// Avoid echoing messages
						continue
					}
					// Pending messages in the inbox
					var pending []*Message
					// Inbox is either empty, in which case
					// new slice will be created, or there's
					// exactly one element with a slice of
					// pending messages, where a new element
					// will be added at the end
					select {
					case pending = <-inbox:
					default:
					}
					// Check for expiration
					if len(pending) > 0 && pending[0].Timestamp.Before(expired) {
						// Client is too slow, but we cannot delete
						// it right now, so schedule it for later
						slow = append(slow, client)
					} else {
						// Overwise append the message and put it
						// into the channel. This is safe and won't
						// block, since we are the only ones writing
						// to it, and it was cleared above
						inbox <- append(pending, message)
					}
				}
				// Discard all slow clients
				for _, client := range(slow) {
					inbox := clients[client]
					delete(clients, client)
					close(inbox)
				}
			}
		}
	}()
	return b
}

func (b *Broker) NewClient() *Client {
	reply := make(chan *Client, 1)
	b.reqadd <- reply
	return <-reply
}

func (b *Broker) RemoveClient(client *Client) {
	b.reqremove <- client
}

var broker = NewBroker()

func main() {
	ready := make(chan struct{})
	go func() {
		client := broker.NewClient()
		defer broker.RemoveClient(client)
		close(ready)
		for batch := range(client.Inbox) {
			for _, message := range(batch) {
				fmt.Printf("%v %s\n", message.Timestamp, message.Value)
			}
			fmt.Printf("...end of batch\n")
		}
		fmt.Printf("...disconnected\n")
	}()
	client := broker.NewClient()
	<-ready
	client.Outbox <- &Message{time.Now(), client, "Hello world!"}
	time.Sleep(time.Nanosecond)
	client.Outbox <- &Message{time.Now(), client, "Are you slow?"}
	time.Sleep(time.Second)
}