I have code that needs to deliver a message (called HintInfo in the sample below) to any number of listeners. These listeners will be attached to web or other clients and thus can be volatile - they can freeze up, time out and disappear, etc. I'm trying to make sure I have a good pattern for how to reliably manage the life cycle on this. What I've come up with is shown below. The basic concept is that a map of channels is kept that corresponds to each of the listeners. Listeners are added or removed atomically using an RWMutex. Sending to these channels is done with a read lock, and a regular synchronous channel send is attempted first. If this doesn't work, a separate goroutine is launched which blocks until the message can be sent to that channel. It is conceivable that thousands of messages could rapidly stack up for a client that is blocked for, let's say a few minutes. I want to make sure this doesn't cause a leak of some sort. The idea here is that when a message comes in, it gets delivered to all channels right away, even if one or more of them is full. Sending is attempted without a separate goroutine first in order to preserve sequence, but when queues fill up later messages will still arrive but possibly out of sequence because a bunch of channel sends in separate goroutines I'm pretty sure have no predictable delivery sequence - this is an acceptable trade-off for my application.
A listener is responsible for ensuring that the channel it used is released (by calling ReleaseRecvChannel() ). This gets a write lock and calls close() on the channel, causing any goroutines that are blocking for a write to panic, which they recover() from and ignore. That should ensure things are properly cleaned up, effectively loosing any undelivered messages for that listener (which is fine, since the listener disconnected, that's the logic course of action).
My question is really just if anyone sees a better way or a problem with this approach. I think this will work, but it's a bit of a tricky problem and I wanted to see if there was something I'm missing.
type HintInfo struct { /* ... */ }
type hintFanout struct {
hintChs map[chan HintInfo]struct{}
l sync.RWMutex
}
func (hf *hintFanout) deliver(hi HintInfo) {
hf.l.RLock()
defer hf.l.RUnlock()
for ch := range hf.hintChs {
// try synchronous send
select {
case ch <- hi:
default:
// can't send right now do it in another goroutine
go func(ch chan HintInfo) {
defer func() { recover() }() // if send panics we ignore it
ch <- hi
}(ch)
}
}
}
func (hf *hintFanout) GetRecvChannel() chan HintInfo {
hf.l.Lock()
defer hf.l.Unlock()
ch := make(chan HintInfo, 4096)
hf.hintChs[ch] = struct{}{}
return ch
}
func (hf *hintFanout) ReleaseRecvChannel(ch chan HintInfo) {
hf.l.Lock()
defer hf.l.Unlock()
// since we have a write lock, there will be no deliver() calls
// in progress right now (only possibly old goroutines trying to
// send to a full channel)
delete(hf.hintChs, ch)
close(ch) // this will cause any goroutines waiting for a channel send to panic
}