func (proxy *PgProxy) ProxyClientConnection(errCh chan<- error) {
	ctx, span := trace.StartSpan(proxy.ctx, "ProxyClientConnection")
	defer span.End()
	logger := logging.NewLoggerWithTrace(ctx).WithField("proxy", "client")
	logger.Debugln("ProxyClientConnection")
	writer := bufio.NewWriter(proxy.dbConnection)
	reader := bufio.NewReader(proxy.clientConnection)
	packet, err := NewClientSidePacketHandler(reader, writer, logger)
	if err != nil {
		logger.WithField(logging.FieldKeyEventCode, logging.EventCodeErrorCodingPostgresqlPacketHandlerInitiailization).WithError(err).Errorln("Can't initialize packet handler object")
		errCh <- err
		return
	}
	prometheusLabels := []string{base.DecryptionDBPostgresql}
	var spanEndFunc = func() {}
	var timerObserveFunc = func() time.Duration { return 0 }
	defer func() {
		spanEndFunc()
		timerObserveFunc()
	}()
	for {
		timerObserveFunc()
		timer := prometheus.NewTimer(prometheus.ObserverFunc(base.RequestProcessingTimeHistogram.WithLabelValues(prometheusLabels...).Observe))
		timerObserveFunc = timer.ObserveDuration
		packet.Reset()
		spanEndFunc()
		packetSpanCtx, packetSpan := trace.StartSpan(ctx, "ProxyClientConnectionLoop")
		spanEndFunc = packetSpan.End
		if err = packet.ReadClientPacket(); err != nil {
			if proxy.tlsSwitch {
				proxy.tlsSwitch = false
				proxy.TLSCh <- true
				return
			}
			logger.WithError(err).Debugln("Can't read packet from client to database")
			errCh <- err
			return
		}
		proxy.dbConnection.SetWriteDeadline(time.Now().Add(network.DefaultNetworkTimeout))
		_, censorSpan := trace.StartSpan(packetSpanCtx, "censor")
		censored, err := proxy.handleClientPacket(packet, logger)
		if err != nil {
			errCh <- err
			return
		}
		censorSpan.End()
		if censored {
			err := proxy.sendClientAcraCensorError(logger)
			if err != nil {
				errCh <- err
				return
			}
			continue
		}
		if err := packet.sendPacket(); err != nil {
			logger.WithField(logging.FieldKeyEventCode, logging.EventCodeErrorNetworkWrite).
				WithError(err).Errorln("Can't send packet")
			errCh <- err
			return
		}
		if packet.terminatePacket {
			errCh <- io.EOF
			return
		}
	}
}
ProxyClientConnection checks every client request using AcraCensor,
if request is allowed, sends it to the Pg database
package postgresql
import (
	"bufio"
	"bytes"
	"context"
	"encoding/binary"
	"errors"
	"io"
	"net"
	"time"
	acracensor "github.com/cossacklabs/acra/acra-censor"
	"github.com/cossacklabs/acra/acra-censor/common"
	"github.com/cossacklabs/acra/decryptor/base"
	"github.com/cossacklabs/acra/logging"
	"github.com/cossacklabs/acra/network"
	"github.com/cossacklabs/acra/sqlparser"
	"github.com/cossacklabs/acra/utils"
	"github.com/prometheus/client_golang/prometheus"
	log "github.com/sirupsen/logrus"
	"go.opencensus.io/trace"
)
var ReadyForQueryPacket = []byte{'Z', 0, 0, 0, 5, 'I'}
var TerminatePacket = []byte{'X', 0, 0, 0, 4}
func NewPgError(message string) ([]byte, error) {
	output := make([]byte, 5+7+7+len(message)+2)
	output[0] = 'E'
	output = output[:5]
	output = append(output, []byte{'S', 'E', 'R', 'R', 'O', 'R', 0}...)
	output = append(output, []byte("C42000")...)
	output = append(output, 0)
	output = append(output, append([]byte{'M'}, []byte(message)...)...)
	output = append(output, 0, 0)
	binary.BigEndian.PutUint32(output[1:5], uint32(len(output)-1))
	return output, nil
}
var (
	ErrInvalidPreparedStatementRegistry = errors.New("ClientSession contains invalid PreparedStatementRegistry")
	ErrInvalidCursorRegistry            = errors.New("ClientSession contains invalid CursorRegistry")
	ErrInvalidProtocolState             = errors.New("ClientSession contains invalid ProtocolState")
)
const (
	DataRowLengthBufSize = 4
	OutputDefaultSize = 1024
	DataRowMessageType       byte = 'D'
	QueryMessageType         byte = 'Q'
	ParseMessageType         byte = 'P'
	BindMessageType          byte = 'B'
	ExecuteMessageType       byte = 'E'
	ParseCompleteMessageType byte = '1'
	BindCompleteMessageType  byte = '2'
	ReadyForQueryMessageType byte = 'Z'
	TLSTimeout                    = time.Second * 2
)
type PgProxy struct {
	session              base.ClientSession
	clientConnection     net.Conn
	dbConnection         net.Conn
	TLSCh                chan bool
	ctx                  context.Context
	queryObserverManager base.QueryObserverManager
	censor               acracensor.AcraCensorInterface
	decryptor            base.Decryptor
	tlsSwitch            bool
	decryptionObserver   base.ColumnDecryptionObserver
	protocolState        *PgProtocolState
	setting              base.ProxySetting
}
func NewPgProxy(session base.ClientSession, decryptor base.Decryptor, setting base.ProxySetting) (*PgProxy, error) {
	observerManager, err := base.NewArrayQueryObserverableManager(session.Context())
	if err != nil {
		return nil, err
	}
	if session.PreparedStatementRegistry() == nil {
		session.SetPreparedStatementRegistry(NewPreparedStatementRegistry())
	}
	var protocolState *PgProtocolState
	if session.ProtocolState() != nil {
		var ok bool
		protocolState, ok = session.ProtocolState().(*PgProtocolState)
		if !ok {
			return nil, ErrInvalidProtocolState
		}
	} else {
		protocolState = NewPgProtocolState()
		session.SetProtocolState(protocolState)
	}
	return &PgProxy{
		session:              session,
		clientConnection:     session.ClientConnection(),
		dbConnection:         session.DatabaseConnection(),
		TLSCh:                make(chan bool),
		ctx:                  session.Context(),
		queryObserverManager: observerManager,
		setting:              setting,
		censor:               setting.Censor(),
		decryptor:            decryptor,
		decryptionObserver:   base.NewColumnDecryptionObserver(),
		protocolState:        protocolState,
	}, nil
}
func (proxy *PgProxy) SubscribeOnColumnDecryption(column int, subscriber base.DecryptionSubscriber) {
	proxy.decryptionObserver.SubscribeOnColumnDecryption(column, subscriber)
}
func (proxy *PgProxy) SubscribeOnAllColumnsDecryption(subscriber base.DecryptionSubscriber) {
	proxy.decryptionObserver.SubscribeOnAllColumnsDecryption(subscriber)
}
func (proxy *PgProxy) Unsubscribe(subscriber base.DecryptionSubscriber) {
	proxy.decryptionObserver.Unsubscribe(subscriber)
}
func (proxy *PgProxy) onColumnDecryption(ctx context.Context, i int, data []byte) ([]byte, error) {
	ctx = base.NewContextWithColumnInfo(ctx, base.NewColumnInfo(i, ""))
	log.WithField("client_id", string(proxy.decryptor.(*PgDecryptor).clientID)).Debugln("Create context")
	ctx = base.NewContextWithClientZoneInfo(ctx, proxy.decryptor.(*PgDecryptor).clientID, proxy.decryptor.GetMatchedZoneID(), proxy.decryptor.IsWithZone())
	return proxy.decryptionObserver.OnColumnDecryption(ctx, i, data)
}
func (proxy *PgProxy) AddQueryObserver(obs base.QueryObserver) {
	proxy.queryObserverManager.AddQueryObserver(obs)
}
func (proxy *PgProxy) RegisteredObserversCount() int {
	return proxy.queryObserverManager.RegisteredObserversCount()
}