func (l *Listener) resync(cn *ListenerConn, notificationChan <-chan *Notification) error {
	doneChan := make(chan error)
	go func() {
		for channel := range l.channels {
			gotResponse, err := cn.Listen(channel)
			if gotResponse && err != nil {
				doneChan <- err
				return
			}
			if err != nil {
				for _ = range notificationChan {
				}
				doneChan <- cn.Err()
				return
			}
		}
		doneChan <- nil
	}()
	for {
		select {
		case _, ok := <-notificationChan:
			if !ok {
				notificationChan = nil
			}
		case err := <-doneChan:
			return err
		}
	}
}
Synchronize the list of channels we want to be listening on with the server
after the connection has been established.
package pq
import (
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)
type Notification struct {
	BePid int
	Channel string
	Extra string
}
func recvNotification(r *readBuf) *Notification {
	bePid := r.int32()
	channel := r.string()
	extra := r.string()
	return &Notification{bePid, channel, extra}
}
const (
	connStateIdle int32 = iota
	connStateExpectResponse
	connStateExpectReadyForQuery
)
type message struct {
	typ byte
	err error
}
var errListenerConnClosed = errors.New("pq: ListenerConn has been closed")
type ListenerConn struct {
	connectionLock sync.Mutex
	cn             *conn
	err            error
	connState int32
	senderLock sync.Mutex
	notificationChan chan<- *Notification
	replyChan chan message
}
func NewListenerConn(name string, notificationChan chan<- *Notification) (*ListenerConn, error) {
	return newDialListenerConn(defaultDialer{}, name, notificationChan)
}
func newDialListenerConn(d Dialer, name string, c chan<- *Notification) (*ListenerConn, error) {
	cn, err := DialOpen(d, name)
	if err != nil {
		return nil, err
	}
	l := &ListenerConn{
		cn:               cn.(*conn),
		notificationChan: c,
		connState:        connStateIdle,
		replyChan:        make(chan message, 2),
	}
	go l.listenerConnMain()
	return l, nil
}
func (l *ListenerConn) acquireSenderLock() error {
	l.senderLock.Lock()
	l.connectionLock.Lock()
	err := l.err
	l.connectionLock.Unlock()
	if err != nil {
		l.senderLock.Unlock()
		return err
	}
	return nil
}
func (l *ListenerConn) releaseSenderLock() {
	l.senderLock.Unlock()
}
func (l *ListenerConn) setState(newState int32) bool {
	var expectedState int32
	switch newState {
	case connStateIdle:
		expectedState = connStateExpectReadyForQuery
	case connStateExpectResponse:
		expectedState = connStateIdle
	case connStateExpectReadyForQuery:
		expectedState = connStateExpectResponse
	default:
		panic(fmt.Sprintf("unexpected listenerConnState %d", newState))
	}
	return atomic.CompareAndSwapInt32(&l.connState, expectedState, newState)
}
func (l *ListenerConn) listenerConnLoop() (err error) {
	defer errRecoverNoErrBadConn(&err)
	r := &readBuf{}
	for {
		t, err := l.cn.recvMessage(r)
		if err != nil {
			return err
		}
		switch t {
		case 'A':
			l.notificationChan <- recvNotification(r)
		case 'T', 'D':
		case 'E':
			if !l.setState(connStateExpectReadyForQuery) {
				return parseError(r)
			}
			l.replyChan <- message{t, parseError(r)}
		case 'C', 'I':
			if !l.setState(connStateExpectReadyForQuery) {
				return fmt.Errorf("unexpected CommandComplete")
			}
		case 'Z':
			if !l.setState(connStateIdle) {
				return fmt.Errorf("unexpected ReadyForQuery")
			}
			l.replyChan <- message{t, nil}
		case 'N', 'S':
		default:
			return fmt.Errorf("unexpected message %q from server in listenerConnLoop", t)
		}
	}
}
func (l *ListenerConn) listenerConnMain() {
	err := l.listenerConnLoop()
	l.connectionLock.Lock()
	if l.err == nil {
		l.err = err
	}
	l.cn.Close()
	l.connectionLock.Unlock()
	close(l.replyChan)
	close(l.notificationChan)
}
func (l *ListenerConn) Listen(channel string) (bool, error) {
	return l.ExecSimpleQuery("LISTEN " + QuoteIdentifier(channel))
}
func (l *ListenerConn) Unlisten(channel string) (bool, error) {
	return l.ExecSimpleQuery("UNLISTEN " + QuoteIdentifier(channel))
}
func (l *ListenerConn) UnlistenAll() (bool, error) {
	return l.ExecSimpleQuery("UNLISTEN *")
}
func (l *ListenerConn) Ping() error {
	sent, err := l.ExecSimpleQuery("")
	if !sent {
		return err
	}
	if err != nil {
		panic(err)
	}
	return nil
}
func (l *ListenerConn) sendSimpleQuery(q string) (err error) {
	defer errRecoverNoErrBadConn(&err)
	if !l.setState(connStateExpectResponse) {
		panic("two queries running at the same time")
	}
	b := &writeBuf{
		buf: []byte("Q\x00\x00\x00\x00"),
		pos: 1,
	}
	b.string(q)
	l.cn.send(b)
	return nil
}
func (l *ListenerConn) ExecSimpleQuery(q string) (executed bool, err error) {
	if err = l.acquireSenderLock(); err != nil {
		return false, err
	}
	defer l.releaseSenderLock()
	err = l.sendSimpleQuery(q)
	if err != nil {
		l.connectionLock.Lock()
		if l.err == nil {
			l.err = err
		}
		l.connectionLock.Unlock()
		l.cn.c.Close()
		return false, err
	}
	for {
		m, ok := <-l.replyChan
		if !ok {
			l.connectionLock.Lock()
			err := l.err
			l.connectionLock.Unlock()
			return false, err
		}
		switch m.typ {
		case 'Z':
			if m.err != nil {
				panic("m.err != nil")
			}
			return true, err
		case 'E':
			if m.err == nil {
				panic("m.err == nil")
			}
			err = m.err
		default:
			return false, fmt.Errorf("unknown response for simple query: %q", m.typ)
		}
	}
}
func (l *ListenerConn) Close() error {
	l.connectionLock.Lock()
	if l.err != nil {
		l.connectionLock.Unlock()
		return errListenerConnClosed
	}
	l.err = errListenerConnClosed
	l.connectionLock.Unlock()
	return l.cn.c.Close()
}
func (l *ListenerConn) Err() error {
	return l.err
}
var errListenerClosed = errors.New("pq: Listener has been closed")
var ErrChannelAlreadyOpen = errors.New("pq: channel is already open")
var ErrChannelNotOpen = errors.New("pq: channel is not open")
type ListenerEventType int
const (
	ListenerEventConnected ListenerEventType = iota
	ListenerEventDisconnected
	ListenerEventReconnected
	ListenerEventConnectionAttemptFailed
)
type EventCallbackType func(event ListenerEventType, err error)
type Listener struct {
	Notify chan *Notification
	name                 string
	minReconnectInterval time.Duration
	maxReconnectInterval time.Duration
	dialer               Dialer
	eventCallback        EventCallbackType
	lock                 sync.Mutex
	isClosed             bool
	reconnectCond        *sync.Cond
	cn                   *ListenerConn
	connNotificationChan <-chan *Notification
	channels             map[string]struct{}
}
func NewListener(name string,
	minReconnectInterval time.Duration,
	maxReconnectInterval time.Duration,
	eventCallback EventCallbackType) *Listener {
	return NewDialListener(defaultDialer{}, name, minReconnectInterval, maxReconnectInterval, eventCallback)
}
func NewDialListener(d Dialer,
	name string,
	minReconnectInterval time.Duration,
	maxReconnectInterval time.Duration,
	eventCallback EventCallbackType) *Listener {
	l := &Listener{
		name:                 name,
		minReconnectInterval: minReconnectInterval,
		maxReconnectInterval: maxReconnectInterval,
		dialer:               d,
		eventCallback:        eventCallback,
		channels: make(map[string]struct{}),
		Notify: make(chan *Notification, 32),
	}
	l.reconnectCond = sync.NewCond(&l.lock)
	go l.listenerMain()
	return l
}
func (l *Listener) NotificationChannel() <-chan *Notification {
	return l.Notify
}
func (l *Listener) Listen(channel string) error {
	l.lock.Lock()
	defer l.lock.Unlock()
	if l.isClosed {
		return errListenerClosed
	}
	_, exists := l.channels[channel]
	if exists {
		return ErrChannelAlreadyOpen
	}
	if l.cn != nil {
		gotResponse, err := l.cn.Listen(channel)
		if gotResponse && err != nil {
			return err
		}
	}
	l.channels[channel] = struct{}{}
	for l.cn == nil {
		l.reconnectCond.Wait()
		if l.isClosed {
			return errListenerClosed
		}
	}
	return nil
}
func (l *Listener) Unlisten(channel string) error {
	l.lock.Lock()
	defer l.lock.Unlock()
	if l.isClosed {
		return errListenerClosed
	}
	_, exists := l.channels[channel]
	if !exists {
		return ErrChannelNotOpen
	}
	if l.cn != nil {
		gotResponse, err := l.cn.Unlisten(channel)
		if gotResponse && err != nil {
			return err
		}
	}
	delete(l.channels, channel)
	return nil
}
func (l *Listener) UnlistenAll() error {
	l.lock.Lock()
	defer l.lock.Unlock()
	if l.isClosed {
		return errListenerClosed
	}
	if l.cn != nil {
		gotResponse, err := l.cn.UnlistenAll()
		if gotResponse && err != nil {
			return err
		}
	}
	l.channels = make(map[string]struct{})
	return nil
}
func (l *Listener) Ping() error {
	l.lock.Lock()
	defer l.lock.Unlock()
	if l.isClosed {
		return errListenerClosed
	}
	if l.cn == nil {
		return errors.New("no connection")
	}
	return l.cn.Ping()
}
func (l *Listener) disconnectCleanup() error {
	l.lock.Lock()
	defer l.lock.Unlock()
	select {
	case _, ok := <-l.connNotificationChan:
		if ok {
			panic("connNotificationChan not closed")
		}
	default:
		panic("connNotificationChan not closed")
	}
	err := l.cn.Err()
	l.cn.Close()
	l.cn = nil
	return err
}