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 'S':
		case 'N':
			if n := l.cn.noticeHandler; n != nil {
				n(parseError(r))
			}
		default:
			return fmt.Errorf("unexpected message %q from server in listenerConnLoop", t)
		}
	}
}
Main logic is here: receive messages from the postgres backend, forward
notifications and query replies and keep the internal state in sync with the
protocol state.  Returns when the connection has been lost, is about to go
away or should be discarded because we couldn't agree on the state with the
server backend.
package pq
import (
	"context"
	"database/sql/driver"
	"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}
}
func SetNotificationHandler(c driver.Conn, handler func(*Notification)) {
	c.(*conn).notificationHandler = handler
}
type NotificationHandlerConnector struct {
	driver.Connector
	notificationHandler func(*Notification)
}
func (n *NotificationHandlerConnector) Connect(ctx context.Context) (driver.Conn, error) {
	c, err := n.Connector.Connect(ctx)
	if err == nil {
		SetNotificationHandler(c, n.notificationHandler)
	}
	return c, err
}
func ConnectorNotificationHandler(c driver.Connector) func(*Notification) {
	if c, ok := c.(*NotificationHandlerConnector); ok {
		return c.notificationHandler
	}
	return nil
}
func ConnectorWithNotificationHandler(c driver.Connector, handler func(*Notification)) *NotificationHandlerConnector {
	if c, ok := c.(*NotificationHandlerConnector); ok {
		c.notificationHandler = handler
		return c
	}
	return &NotificationHandlerConnector{Connector: c, notificationHandler: handler}
}
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)
}