func (d *NotifyDispatcher) execListen(channel string) {
	err := d.listener.Listen(channel)
	if err == pq.ErrChannelAlreadyOpen {
		err = nil
	}
	d.lock.Lock()
	defer d.lock.Unlock()
	set, ok := d.channels[channel]
	if !ok {
		panic("oops")
	}
	if err != nil {
		delete(d.channels, channel)
	}
	set.markActive(err)
}
Attempt to start listening on channel.  The caller should not be holding
lock.
package notifydispatcher
import (
	"container/list"
	"errors"
	"fmt"
	"github.com/lib/pq"
	"sync"
)
var (
	ErrChannelAlreadyActive = errors.New("channel is already active")
	ErrChannelNotActive = errors.New("channel is not active")
)
var errClosed = errors.New("NotifyDispatcher has been closed")
type SlowReaderEliminationStrategy int
const (
	CloseSlowReaders SlowReaderEliminationStrategy = iota
	NeglectSlowReaders
)
type listenRequest struct {
	channel string
	unlisten bool
}
type BroadcastChannel struct {
	Channel chan struct{}
	elem *list.Element
}
type Listener interface {
	Listen(channel string) error
	Unlisten(channel string) error
	NotificationChannel() <-chan *pq.Notification
}
type NotifyDispatcher struct {
	listener Listener
	slowReaderEliminationStrategy SlowReaderEliminationStrategy
	broadcastOnConnectionLoss bool
	broadcastChannels *list.List
	listenRequestch chan listenRequest
	lock sync.Mutex
	channels map[string] *listenSet
	closed bool
	closeChannel chan bool
}
func NewNotifyDispatcher(l Listener) *NotifyDispatcher {
	d := &NotifyDispatcher{
		listener: l,
		slowReaderEliminationStrategy: CloseSlowReaders,
		broadcastOnConnectionLoss: true,
		broadcastChannels: list.New(),
		listenRequestch: make(chan listenRequest, 64),
		channels: make(map[string] *listenSet),
		closeChannel: make(chan bool),
	}
	go d.dispatcherLoop()
	go d.listenRequestHandlerLoop()
	return d
}
func (d *NotifyDispatcher) NumActiveChannels() int {
	d.lock.Lock()
	numActiveChannels := len(d.channels)
	d.lock.Unlock()
	return numActiveChannels
}
func (d *NotifyDispatcher) SetSlowReaderEliminationStrategy(strategy SlowReaderEliminationStrategy) {
	d.lock.Lock()
	defer d.lock.Unlock()
	d.slowReaderEliminationStrategy = strategy
}
func (d *NotifyDispatcher) SetBroadcastOnConnectionLoss(value bool) {
	d.lock.Lock()
	defer d.lock.Unlock()
	d.broadcastOnConnectionLoss = value
}
func (d *NotifyDispatcher) OpenBroadcastChannel() BroadcastChannel {
	d.lock.Lock()
	defer d.lock.Unlock()
	ch := BroadcastChannel{
		Channel: make(chan struct{}, 1),
	}
	ch.elem = d.broadcastChannels.PushFront(ch)
	return ch
}
func (d *NotifyDispatcher) CloseBroadcastChannel(ch BroadcastChannel) {
	d.lock.Lock()
	defer d.lock.Unlock()
	r := d.broadcastChannels.Remove(ch.elem).(BroadcastChannel)
	if r.Channel != ch.Channel {
		panic("oops")
	}
	close(ch.Channel)
}
func (d *NotifyDispatcher) broadcast() {
	d.lock.Lock()
	for e := d.broadcastChannels.Front(); e != nil; e = e.Next() {
		select {
			case e.Value.(BroadcastChannel).Channel <- struct{}{}:
			default:
		}
	}
	if !d.broadcastOnConnectionLoss {
		d.lock.Unlock()
		return
	}
	var reapchans []string
	for channel, set := range d.channels {
		if !set.broadcast(d.slowReaderEliminationStrategy, nil) {
			reapchans = append(reapchans, channel)
		}
	}
	d.lock.Unlock()
	for _, ch := range reapchans {
		err := d.requestListen(ch, true)
		if err != nil {
			if err != errClosed {
				panic(err)
			}
			break
		}
	}
}
func (d *NotifyDispatcher) dispatch(n *pq.Notification) {
	reap := false
	d.lock.Lock()
	set, ok := d.channels[n.Channel]
	if ok {
		reap = !set.broadcast(d.slowReaderEliminationStrategy, n)
	}
	d.lock.Unlock()
	if reap {
		err := d.requestListen(n.Channel, true)
		if err != nil && err != errClosed {
			panic(err)
		}
	}
}
func (d *NotifyDispatcher) dispatcherLoop() {
	notifyCh := d.listener.NotificationChannel()
dispatcherLoop:
	for {
		var n *pq.Notification
		select {
			case <-d.closeChannel:
				break dispatcherLoop
			case n = <-notifyCh:
		}
		if n == nil {
			d.broadcast()
		} else {
			d.dispatch(n)
		}
	}
}