a

lib/gat/pools/session/pool.go

@@ -53,9 +53,7 @@ func (T *Pool) Serve(client zap.ReadWriter) {
 	server := T.acquire()
 	defer T.release(server)
 	for {
-		if err := client.Poll(); err != nil {
-			break
-		}
+		// TODO(garet) test if client has disconnected
 		bouncers.Bounce(client, server)
 	}
 }

lib/gat/pools/transaction/pool.go

@@ -10,6 +10,7 @@ import (
 	"pggat2/lib/middleware/middlewares/ps"
 	"pggat2/lib/rob/schedulers/v1"
 	"pggat2/lib/zap"
+	"pggat2/lib/zap/zapbuf"
 )
 
 type Pool struct {
@@ -66,12 +67,14 @@ func (T *Pool) Serve(client zap.ReadWriter) {
 		eqpc,
 		psc,
 	)
+	buffer := zapbuf.NewBuffer(client)
+	defer buffer.Done()
 	for {
-		if err := client.Poll(); err != nil {
+		if err := buffer.Buffer(); err != nil {
 			break
 		}
 		source.Do(0, Work{
-			rw:  client,
+			rw:  buffer,
 			eqp: eqpc,
 			ps:  psc,
 		})

lib/middleware/interceptor/interceptor.go

@@ -23,10 +23,6 @@ func NewInterceptor(rw zap.ReadWriter, middlewares ...middleware.Middleware) *In
 	}
 }
 
-func (T *Interceptor) Poll() error {
-	return T.rw.Poll()
-}
-
 func (T *Interceptor) ReadByte() (byte, error) {
 	return T.rw.ReadByte()
 }

lib/zap/reader.go

@@ -3,7 +3,6 @@ package zap
 import "io"
 
 type Reader interface {
-	Poll() error
 	ReadByte() (byte, error)
 	Read(*Packet) error
 	ReadUntyped(*UntypedPacket) error
@@ -13,11 +12,6 @@ type IOReader struct {
 	Reader io.Reader
 }
 
-func (T IOReader) Poll() error {
-	_, err := T.Reader.Read([]byte{})
-	return err
-}
-
 func (T IOReader) ReadByte() (byte, error) {
 	var res = []byte{0}
 	_, err := io.ReadFull(T.Reader, res)

lib/zap/zapbuf/readwriter.go

+package zapbuf
+
+import (
+	"errors"
+
+	"pggat2/lib/zap"
+)
+
+var ErrAlreadyBuffered = errors.New("already buffered")
+
+type Buffer struct {
+	zap.ReadWriter
+	buffered  *zap.Packet
+	hasBuffer bool
+}
+
+func NewBuffer(rw zap.ReadWriter) *Buffer {
+	return &Buffer{
+		ReadWriter: rw,
+	}
+}
+
+func (T *Buffer) Buffer() error {
+	if T.hasBuffer {
+		return ErrAlreadyBuffered
+	}
+
+	T.hasBuffer = true
+
+	if T.buffered == nil {
+		T.buffered = zap.NewPacket()
+	}
+
+	return T.ReadWriter.Read(T.buffered)
+}
+
+func (T *Buffer) Read(packet *zap.Packet) error {
+	if T.hasBuffer {
+		// swap buffers
+		packet.PacketWriter, T.buffered.PacketWriter = T.buffered.PacketWriter, packet.PacketWriter
+		T.hasBuffer = false
+		return nil
+	}
+
+	return T.ReadWriter.Read(packet)
+}
+
+func (T *Buffer) Done() {
+	if T.buffered != nil {
+		T.buffered.Done()
+	}
+}
+
+var _ zap.ReadWriter = (*Buffer)(nil)