From 55d7f7e13daafdfb1ed21de87f4aafa543c97ede Mon Sep 17 00:00:00 2001
From: Garet Halliday <me@garet.holiday>
Date: Thu, 28 Sep 2023 16:14:00 -0500
Subject: [PATCH] slightly better
---
lib/fed/conn.go | 204 +++---------------
lib/fed/middleware.go | 7 +
lib/fed/middlewares/eqp/client.go | 25 +++
lib/fed/middlewares/eqp/server.go | 25 +++
.../middlewares/eqp/state.go | 0
.../middlewares/eqp/sync.go | 0
.../middlewares/ps/client.go | 16 +-
.../middlewares/ps/server.go | 13 +-
.../middlewares/ps/sync.go | 0
.../middlewares/unterminate/unterminate.go | 13 +-
lib/fed/netconn.go | 144 +++++++++++++
lib/fed/readwriter.go | 7 +-
lib/fed/ssl.go | 8 +
lib/gat/app.go | 30 ++-
lib/gat/listen.go | 6 +-
lib/gat/matcher.go | 2 +-
lib/gat/matchers/and.go | 2 +-
lib/gat/matchers/database.go | 4 +-
lib/gat/matchers/localaddress.go | 8 +-
lib/gat/matchers/or.go | 2 +-
lib/gat/matchers/ssl.go | 8 +-
lib/gat/matchers/startupparameters.go | 5 +-
lib/gat/matchers/user.go | 4 +-
lib/gat/pool/client.go | 26 +--
lib/gat/pool/{options.go => config.go} | 0
lib/gat/pool/conn.go | 18 +-
lib/gat/pool/flow.go | 10 +-
lib/gat/pool/pool.go | 12 +-
lib/gat/pool/recipe/{options.go => config.go} | 0
lib/gat/pool/recipe/dialer.go | 18 +-
lib/gat/pool/recipe/recipe.go | 2 +-
lib/gat/pool/server.go | 23 +-
lib/gat/provider.go | 2 +-
.../google_cloud_sql/discoverer.go | 2 +-
lib/gat/providers/discovery/module.go | 4 +-
lib/gat/providers/pgbouncer/module.go | 4 +-
lib/gat/server.go | 2 +-
lib/gsql/client.go | 31 +--
lib/middleware/context.go | 11 -
lib/middleware/interceptor/context.go | 36 ----
lib/middleware/interceptor/interceptor.go | 100 ---------
lib/middleware/middleware.go | 8 -
lib/middleware/middlewares/eqp/client.go | 26 ---
lib/middleware/middlewares/eqp/server.go | 26 ---
44 files changed, 357 insertions(+), 537 deletions(-)
create mode 100644 lib/fed/middleware.go
create mode 100644 lib/fed/middlewares/eqp/client.go
create mode 100644 lib/fed/middlewares/eqp/server.go
rename lib/{middleware => fed}/middlewares/eqp/state.go (100%)
rename lib/{middleware => fed}/middlewares/eqp/sync.go (100%)
rename lib/{middleware => fed}/middlewares/ps/client.go (69%)
rename lib/{middleware => fed}/middlewares/ps/server.go (68%)
rename lib/{middleware => fed}/middlewares/ps/sync.go (100%)
rename lib/{middleware => fed}/middlewares/unterminate/unterminate.go (60%)
create mode 100644 lib/fed/netconn.go
rename lib/gat/pool/{options.go => config.go} (100%)
rename lib/gat/pool/recipe/{options.go => config.go} (100%)
delete mode 100644 lib/middleware/context.go
delete mode 100644 lib/middleware/interceptor/context.go
delete mode 100644 lib/middleware/interceptor/interceptor.go
delete mode 100644 lib/middleware/middleware.go
delete mode 100644 lib/middleware/middlewares/eqp/client.go
delete mode 100644 lib/middleware/middlewares/eqp/server.go
diff --git a/lib/fed/conn.go b/lib/fed/conn.go
index 4f27d45a..b82955b3 100644
--- a/lib/fed/conn.go
+++ b/lib/fed/conn.go
@@ -1,195 +1,61 @@
package fed
import (
- "bufio"
- "crypto/tls"
- "encoding/binary"
- "errors"
- "io"
- "net"
-
- "gfx.cafe/gfx/pggat/lib/util/slices"
+ "gfx.cafe/gfx/pggat/lib/util/decorator"
"gfx.cafe/gfx/pggat/lib/util/strutil"
)
-type Conn interface {
- ReadWriter
-
- LocalAddr() net.Addr
- RemoteAddr() net.Addr
-
- SSLEnabled() bool
- User() string
- Database() string
- InitialParameters() map[strutil.CIString]string
- BackendKey() [8]byte
-
- Close() error
-}
-
-type NetConn struct {
- conn net.Conn
- writer bufio.Writer
- reader bufio.Reader
- sslEnabled bool
-
- user string
- database string
- initialParameters map[strutil.CIString]string
- backendKey [8]byte
-
- headerBuf [5]byte
-}
-
-func WrapNetConn(conn net.Conn) *NetConn {
- c := &NetConn{
- conn: conn,
- }
- c.writer.Reset(conn)
- c.reader.Reset(conn)
- return c
-}
-
-func (T *NetConn) LocalAddr() net.Addr {
- return T.conn.LocalAddr()
-}
-
-func (T *NetConn) RemoteAddr() net.Addr {
- return T.conn.RemoteAddr()
-}
-
-func (T *NetConn) SSLEnabled() bool {
- return T.sslEnabled
-}
-
-func (T *NetConn) User() string {
- return T.user
-}
-
-func (T *NetConn) SetUser(user string) {
- T.user = user
-}
-
-func (T *NetConn) Database() string {
- return T.database
-}
-
-func (T *NetConn) SetDatabase(database string) {
- T.database = database
-}
-
-func (T *NetConn) InitialParameters() map[strutil.CIString]string {
- return T.initialParameters
-}
-
-func (T *NetConn) SetInitialParameters(initialParameters map[strutil.CIString]string) {
- T.initialParameters = initialParameters
-}
-
-func (T *NetConn) BackendKey() [8]byte {
- return T.backendKey
-}
-
-func (T *NetConn) SetBackendKey(backendKey [8]byte) {
- T.backendKey = backendKey
-}
+type Conn struct {
+ noCopy decorator.NoCopy
-var errSSLAlreadyEnabled = errors.New("ssl is already enabled")
+ ReadWriteCloser
-func (T *NetConn) EnableSSLClient(config *tls.Config) error {
- if T.sslEnabled {
- return errSSLAlreadyEnabled
- }
- T.sslEnabled = true
+ Middleware []Middleware
- if err := T.writer.Flush(); err != nil {
- return err
- }
- if T.reader.Buffered() > 0 {
- return errors.New("expected empty read buffer")
- }
- sslConn := tls.Client(T.conn, config)
- T.writer.Reset(sslConn)
- T.reader.Reset(sslConn)
- T.conn = sslConn
- return sslConn.Handshake()
+ User string
+ Database string
+ InitialParameters map[strutil.CIString]string
+ Authenticated bool
+ BackendKey [8]byte
}
-func (T *NetConn) EnableSSLServer(config *tls.Config) error {
- if T.sslEnabled {
- return errSSLAlreadyEnabled
+func NewConn(rw ReadWriteCloser) *Conn {
+ return &Conn{
+ ReadWriteCloser: rw,
}
- T.sslEnabled = true
-
- if err := T.writer.Flush(); err != nil {
- return err
- }
- if T.reader.Buffered() > 0 {
- return errors.New("expected empty read buffer")
- }
- sslConn := tls.Server(T.conn, config)
- T.writer.Reset(sslConn)
- T.reader.Reset(sslConn)
- T.conn = sslConn
- return sslConn.Handshake()
}
-func (T *NetConn) ReadByte() (byte, error) {
- if err := T.writer.Flush(); err != nil {
- return 0, err
- }
- return T.reader.ReadByte()
-}
-
-func (T *NetConn) ReadPacket(typed bool, buffer Packet) (packet Packet, err error) {
+func (T *Conn) ReadPacket(typed bool, buffer Packet) (packet Packet, err error) {
packet = buffer
-
- if err = T.writer.Flush(); err != nil {
- return
- }
-
- if typed {
- _, err = io.ReadFull(&T.reader, T.headerBuf[:])
+ for {
+ packet, err = T.ReadWriteCloser.ReadPacket(typed, buffer)
if err != nil {
return
}
- } else {
- _, err = io.ReadFull(&T.reader, T.headerBuf[1:])
- if err != nil {
+ for _, middleware := range T.Middleware {
+ packet, err = middleware.ReadPacket(packet)
+ if err != nil {
+ return
+ }
+ if len(packet) == 0 {
+ break
+ }
+ }
+ if len(packet) != 0 {
return
}
}
-
- length := binary.BigEndian.Uint32(T.headerBuf[1:])
-
- packet = slices.Resize(buffer, int(length)+1)
- copy(packet, T.headerBuf[:])
-
- _, err = io.ReadFull(&T.reader, packet.Payload())
- if err != nil {
- return
- }
- return
}
-func (T *NetConn) WriteByte(b byte) error {
- return T.writer.WriteByte(b)
-}
-
-func (T *NetConn) WritePacket(packet Packet) error {
- _, err := T.writer.Write(packet.Bytes())
- return err
-}
-
-func (T *NetConn) Close() error {
- if err := T.writer.Flush(); err != nil {
- return err
+func (T *Conn) WritePacket(packet Packet) (err error) {
+ for _, middleware := range T.Middleware {
+ packet, err = middleware.ReadPacket(packet)
+ if err != nil || len(packet) == 0 {
+ return
+ }
}
- return T.conn.Close()
+ err = T.ReadWriteCloser.WritePacket(packet)
+ return
}
-var _ Conn = (*NetConn)(nil)
-var _ SSLServer = (*NetConn)(nil)
-var _ SSLClient = (*NetConn)(nil)
-var _ io.ByteReader = (*NetConn)(nil)
-var _ io.ByteWriter = (*NetConn)(nil)
+var _ ReadWriteCloser = (*Conn)(nil)
diff --git a/lib/fed/middleware.go b/lib/fed/middleware.go
new file mode 100644
index 00000000..c47a3f70
--- /dev/null
+++ b/lib/fed/middleware.go
@@ -0,0 +1,7 @@
+package fed
+
+// Middleware intercepts packets and possibly changes them. Return a 0 length packet to cancel.
+type Middleware interface {
+ ReadPacket(packet Packet) (Packet, error)
+ WritePacket(packet Packet) (Packet, error)
+}
diff --git a/lib/fed/middlewares/eqp/client.go b/lib/fed/middlewares/eqp/client.go
new file mode 100644
index 00000000..47417442
--- /dev/null
+++ b/lib/fed/middlewares/eqp/client.go
@@ -0,0 +1,25 @@
+package eqp
+
+import (
+ "gfx.cafe/gfx/pggat/lib/fed"
+)
+
+type Client struct {
+ state State
+}
+
+func NewClient() *Client {
+ return new(Client)
+}
+
+func (T *Client) ReadPacket(packet fed.Packet) (fed.Packet, error) {
+ T.state.C2S(packet)
+ return packet, nil
+}
+
+func (T *Client) WritePacket(packet fed.Packet) (fed.Packet, error) {
+ T.state.S2C(packet)
+ return packet, nil
+}
+
+var _ fed.Middleware = (*Client)(nil)
diff --git a/lib/fed/middlewares/eqp/server.go b/lib/fed/middlewares/eqp/server.go
new file mode 100644
index 00000000..f1b9e788
--- /dev/null
+++ b/lib/fed/middlewares/eqp/server.go
@@ -0,0 +1,25 @@
+package eqp
+
+import (
+ "gfx.cafe/gfx/pggat/lib/fed"
+)
+
+type Server struct {
+ state State
+}
+
+func NewServer() *Server {
+ return new(Server)
+}
+
+func (T *Server) ReadPacket(packet fed.Packet) (fed.Packet, error) {
+ T.state.S2C(packet)
+ return packet, nil
+}
+
+func (T *Server) WritePacket(packet fed.Packet) (fed.Packet, error) {
+ T.state.C2S(packet)
+ return packet, nil
+}
+
+var _ fed.Middleware = (*Server)(nil)
diff --git a/lib/middleware/middlewares/eqp/state.go b/lib/fed/middlewares/eqp/state.go
similarity index 100%
rename from lib/middleware/middlewares/eqp/state.go
rename to lib/fed/middlewares/eqp/state.go
diff --git a/lib/middleware/middlewares/eqp/sync.go b/lib/fed/middlewares/eqp/sync.go
similarity index 100%
rename from lib/middleware/middlewares/eqp/sync.go
rename to lib/fed/middlewares/eqp/sync.go
diff --git a/lib/middleware/middlewares/ps/client.go b/lib/fed/middlewares/ps/client.go
similarity index 69%
rename from lib/middleware/middlewares/ps/client.go
rename to lib/fed/middlewares/ps/client.go
index c0b4fca6..ecb819f9 100644
--- a/lib/middleware/middlewares/ps/client.go
+++ b/lib/fed/middlewares/ps/client.go
@@ -5,7 +5,6 @@ import (
"gfx.cafe/gfx/pggat/lib/fed"
packets "gfx.cafe/gfx/pggat/lib/fed/packets/v3.0"
- "gfx.cafe/gfx/pggat/lib/middleware"
"gfx.cafe/gfx/pggat/lib/util/strutil"
)
@@ -20,29 +19,28 @@ func NewClient(parameters map[strutil.CIString]string) *Client {
}
}
-func (T *Client) Read(_ middleware.Context, _ fed.Packet) error {
- return nil
+func (T *Client) ReadPacket(packet fed.Packet) (fed.Packet, error) {
+ return packet, nil
}
-func (T *Client) Write(ctx middleware.Context, packet fed.Packet) error {
+func (T *Client) WritePacket(packet fed.Packet) (fed.Packet, error) {
switch packet.Type() {
case packets.TypeParameterStatus:
var ps packets.ParameterStatus
if !ps.ReadFromPacket(packet) {
- return errors.New("bad packet format i")
+ return packet, errors.New("bad packet format i")
}
ikey := strutil.MakeCIString(ps.Key)
if T.synced && T.parameters[ikey] == ps.Value {
// already set
- ctx.Cancel()
- break
+ return packet[:0], nil
}
if T.parameters == nil {
T.parameters = make(map[strutil.CIString]string)
}
T.parameters[ikey] = ps.Value
}
- return nil
+ return packet, nil
}
-var _ middleware.Middleware = (*Client)(nil)
+var _ fed.Middleware = (*Client)(nil)
diff --git a/lib/middleware/middlewares/ps/server.go b/lib/fed/middlewares/ps/server.go
similarity index 68%
rename from lib/middleware/middlewares/ps/server.go
rename to lib/fed/middlewares/ps/server.go
index dc11a6e1..f74f4f0c 100644
--- a/lib/middleware/middlewares/ps/server.go
+++ b/lib/fed/middlewares/ps/server.go
@@ -5,7 +5,6 @@ import (
"gfx.cafe/gfx/pggat/lib/fed"
packets "gfx.cafe/gfx/pggat/lib/fed/packets/v3.0"
- "gfx.cafe/gfx/pggat/lib/middleware"
"gfx.cafe/gfx/pggat/lib/util/strutil"
)
@@ -19,12 +18,12 @@ func NewServer(parameters map[strutil.CIString]string) *Server {
}
}
-func (T *Server) Read(_ middleware.Context, packet fed.Packet) error {
+func (T *Server) ReadPacket(packet fed.Packet) (fed.Packet, error) {
switch packet.Type() {
case packets.TypeParameterStatus:
var ps packets.ParameterStatus
if !ps.ReadFromPacket(packet) {
- return errors.New("bad packet format j")
+ return packet, errors.New("bad packet format j")
}
ikey := strutil.MakeCIString(ps.Key)
if T.parameters == nil {
@@ -32,11 +31,11 @@ func (T *Server) Read(_ middleware.Context, packet fed.Packet) error {
}
T.parameters[ikey] = ps.Value
}
- return nil
+ return packet, nil
}
-func (T *Server) Write(_ middleware.Context, _ fed.Packet) error {
- return nil
+func (T *Server) WritePacket(packet fed.Packet) (fed.Packet, error) {
+ return packet, nil
}
-var _ middleware.Middleware = (*Server)(nil)
+var _ fed.Middleware = (*Server)(nil)
diff --git a/lib/middleware/middlewares/ps/sync.go b/lib/fed/middlewares/ps/sync.go
similarity index 100%
rename from lib/middleware/middlewares/ps/sync.go
rename to lib/fed/middlewares/ps/sync.go
diff --git a/lib/middleware/middlewares/unterminate/unterminate.go b/lib/fed/middlewares/unterminate/unterminate.go
similarity index 60%
rename from lib/middleware/middlewares/unterminate/unterminate.go
rename to lib/fed/middlewares/unterminate/unterminate.go
index dcd27cfd..f04fb06b 100644
--- a/lib/middleware/middlewares/unterminate/unterminate.go
+++ b/lib/fed/middlewares/unterminate/unterminate.go
@@ -5,7 +5,6 @@ import (
"gfx.cafe/gfx/pggat/lib/fed"
packets "gfx.cafe/gfx/pggat/lib/fed/packets/v3.0"
- "gfx.cafe/gfx/pggat/lib/middleware"
)
// Unterminate catches the Terminate packet and returns io.EOF instead.
@@ -14,15 +13,15 @@ var Unterminate = unterm{}
type unterm struct{}
-func (unterm) Read(_ middleware.Context, packet fed.Packet) error {
+func (unterm) ReadPacket(packet fed.Packet) (fed.Packet, error) {
if packet.Type() == packets.TypeTerminate {
- return io.EOF
+ return packet, io.EOF
}
- return nil
+ return packet, nil
}
-func (unterm) Write(_ middleware.Context, _ fed.Packet) error {
- return nil
+func (unterm) WritePacket(packet fed.Packet) (fed.Packet, error) {
+ return packet, nil
}
-var _ middleware.Middleware = unterm{}
+var _ fed.Middleware = unterm{}
diff --git a/lib/fed/netconn.go b/lib/fed/netconn.go
new file mode 100644
index 00000000..480edb12
--- /dev/null
+++ b/lib/fed/netconn.go
@@ -0,0 +1,144 @@
+package fed
+
+import (
+ "bufio"
+ "crypto/tls"
+ "encoding/binary"
+ "errors"
+ "io"
+ "net"
+
+ "gfx.cafe/gfx/pggat/lib/util/slices"
+)
+
+type NetConn struct {
+ conn net.Conn
+ writer bufio.Writer
+ reader bufio.Reader
+ sslEnabled bool
+
+ headerBuf [5]byte
+}
+
+func NewNetConn(conn net.Conn) *NetConn {
+ c := &NetConn{
+ conn: conn,
+ }
+ c.writer.Reset(conn)
+ c.reader.Reset(conn)
+ return c
+}
+
+func (T *NetConn) LocalAddr() net.Addr {
+ return T.conn.LocalAddr()
+}
+
+func (T *NetConn) RemoteAddr() net.Addr {
+ return T.conn.RemoteAddr()
+}
+
+// SSL
+
+var errSSLAlreadyEnabled = errors.New("ssl is already enabled")
+
+func (T *NetConn) SSL() bool {
+ return T.sslEnabled
+}
+
+func (T *NetConn) EnableSSLClient(config *tls.Config) error {
+ if T.sslEnabled {
+ return errSSLAlreadyEnabled
+ }
+ T.sslEnabled = true
+
+ if err := T.writer.Flush(); err != nil {
+ return err
+ }
+ if T.reader.Buffered() > 0 {
+ return errors.New("expected empty read buffer")
+ }
+ sslConn := tls.Client(T.conn, config)
+ T.writer.Reset(sslConn)
+ T.reader.Reset(sslConn)
+ T.conn = sslConn
+ return sslConn.Handshake()
+}
+
+func (T *NetConn) EnableSSLServer(config *tls.Config) error {
+ if T.sslEnabled {
+ return errSSLAlreadyEnabled
+ }
+ T.sslEnabled = true
+
+ if err := T.writer.Flush(); err != nil {
+ return err
+ }
+ if T.reader.Buffered() > 0 {
+ return errors.New("expected empty read buffer")
+ }
+ sslConn := tls.Server(T.conn, config)
+ T.writer.Reset(sslConn)
+ T.reader.Reset(sslConn)
+ T.conn = sslConn
+ return sslConn.Handshake()
+}
+
+func (T *NetConn) ReadByte() (byte, error) {
+ if err := T.writer.Flush(); err != nil {
+ return 0, err
+ }
+ return T.reader.ReadByte()
+}
+
+func (T *NetConn) ReadPacket(typed bool, buffer Packet) (packet Packet, err error) {
+ packet = buffer
+
+ if err = T.writer.Flush(); err != nil {
+ return
+ }
+
+ if typed {
+ _, err = io.ReadFull(&T.reader, T.headerBuf[:])
+ if err != nil {
+ return
+ }
+ } else {
+ _, err = io.ReadFull(&T.reader, T.headerBuf[1:])
+ if err != nil {
+ return
+ }
+ }
+
+ length := binary.BigEndian.Uint32(T.headerBuf[1:])
+
+ packet = slices.Resize(buffer, int(length)+1)
+ copy(packet, T.headerBuf[:])
+
+ _, err = io.ReadFull(&T.reader, packet.Payload())
+ if err != nil {
+ return
+ }
+ return
+}
+
+func (T *NetConn) WriteByte(b byte) error {
+ return T.writer.WriteByte(b)
+}
+
+func (T *NetConn) WritePacket(packet Packet) error {
+ _, err := T.writer.Write(packet.Bytes())
+ return err
+}
+
+func (T *NetConn) Close() error {
+ if err := T.writer.Flush(); err != nil {
+ return err
+ }
+ return T.conn.Close()
+}
+
+var _ ReadWriteCloser = (*NetConn)(nil)
+var _ SSLServer = (*NetConn)(nil)
+var _ SSLClient = (*NetConn)(nil)
+var _ io.ByteReader = (*NetConn)(nil)
+var _ io.ByteWriter = (*NetConn)(nil)
diff --git a/lib/fed/readwriter.go b/lib/fed/readwriter.go
index e26d66f9..bde9a0d1 100644
--- a/lib/fed/readwriter.go
+++ b/lib/fed/readwriter.go
@@ -13,7 +13,8 @@ type ReadWriter interface {
Writer
}
-type CombinedReadWriter struct {
- Reader
- Writer
+type ReadWriteCloser interface {
+ ReadWriter
+
+ Close() error
}
diff --git a/lib/fed/ssl.go b/lib/fed/ssl.go
index 580d83f9..9b830d23 100644
--- a/lib/fed/ssl.go
+++ b/lib/fed/ssl.go
@@ -2,10 +2,18 @@ package fed
import "crypto/tls"
+type SSL interface {
+ SSL() bool
+}
+
type SSLClient interface {
+ SSL
+
EnableSSLClient(config *tls.Config) error
}
type SSLServer interface {
+ SSL
+
EnableSSLServer(config *tls.Config) error
}
diff --git a/lib/gat/app.go b/lib/gat/app.go
index 213cf39c..b8be6941 100644
--- a/lib/gat/app.go
+++ b/lib/gat/app.go
@@ -90,9 +90,8 @@ func (T *App) cancel(key [8]byte) {
_ = p.Cancel(key)
}
-func (T *App) serve(server *Server, conn *fed.NetConn) {
- initialParameters := conn.InitialParameters()
- for key := range initialParameters {
+func (T *App) serve(server *Server, conn *fed.Conn) {
+ for key := range conn.InitialParameters {
if !slices.Contains(server.AllowedStartupParameters, key) {
errResp := packets.ErrorResponse{
Error: perror.New(
@@ -108,20 +107,20 @@ func (T *App) serve(server *Server, conn *fed.NetConn) {
p := server.lookup(conn)
if p == nil {
- T.log.Warn("database not found", zap.String("user", conn.User()), zap.String("database", conn.Database()))
+ T.log.Warn("database not found", zap.String("user", conn.User), zap.String("database", conn.Database))
return
}
- backendKey, err := frontends.Authenticate(conn, p.Credentials())
+ var err error
+ conn.BackendKey, err = frontends.Authenticate(conn.ReadWriteCloser, p.Credentials())
if err != nil {
T.log.Warn("error authenticating client", zap.Error(err))
return
}
+ conn.Authenticated = true
- conn.SetBackendKey(backendKey)
-
- T.keys.Store(backendKey, p)
- defer T.keys.Delete(backendKey)
+ T.keys.Store(conn.BackendKey, p)
+ defer T.keys.Delete(conn.BackendKey)
if err2 := p.Serve(conn); err2 != nil && !errors.Is(err2, io.EOF) {
T.log.Warn("error serving client", zap.Error(err2))
@@ -129,7 +128,7 @@ func (T *App) serve(server *Server, conn *fed.NetConn) {
}
}
-func (T *App) accept(listener *Listener, conn *fed.NetConn) {
+func (T *App) accept(listener *Listener, conn *fed.Conn) {
defer func() {
_ = conn.Close()
}()
@@ -139,7 +138,10 @@ func (T *App) accept(listener *Listener, conn *fed.NetConn) {
tlsConfig = listener.ssl.ServerTLSConfig()
}
- cancelKey, isCanceling, _, user, database, initialParameters, err := frontends.Accept(conn, tlsConfig)
+ var cancelKey [8]byte
+ var isCanceling bool
+ var err error
+ cancelKey, isCanceling, _, conn.User, conn.Database, conn.InitialParameters, err = frontends.Accept(conn.ReadWriteCloser, tlsConfig)
if err != nil {
T.log.Warn("error accepting client", zap.Error(err))
return
@@ -150,10 +152,6 @@ func (T *App) accept(listener *Listener, conn *fed.NetConn) {
return
}
- conn.SetUser(user)
- conn.SetDatabase(database)
- conn.SetInitialParameters(initialParameters)
-
for _, server := range T.servers {
if server.match == nil || server.match.Matches(conn) {
T.serve(server, conn)
@@ -161,7 +159,7 @@ func (T *App) accept(listener *Listener, conn *fed.NetConn) {
}
}
- T.log.Warn("server not found", zap.String("user", conn.User()), zap.String("database", conn.Database()))
+ T.log.Warn("server not found", zap.String("user", conn.User), zap.String("database", conn.Database))
errResp := packets.ErrorResponse{
Error: perror.New(
diff --git a/lib/gat/listen.go b/lib/gat/listen.go
index dd0f2866..deca2667 100644
--- a/lib/gat/listen.go
+++ b/lib/gat/listen.go
@@ -27,12 +27,14 @@ type Listener struct {
log *zap.Logger
}
-func (T *Listener) accept() (*fed.NetConn, error) {
+func (T *Listener) accept() (*fed.Conn, error) {
raw, err := T.listener.Accept()
if err != nil {
return nil, err
}
- return fed.WrapNetConn(raw), nil
+ return fed.NewConn(
+ fed.NewNetConn(raw),
+ ), nil
}
func (T *Listener) Provision(ctx caddy.Context) error {
diff --git a/lib/gat/matcher.go b/lib/gat/matcher.go
index 20c20618..6152f4e4 100644
--- a/lib/gat/matcher.go
+++ b/lib/gat/matcher.go
@@ -3,5 +3,5 @@ package gat
import "gfx.cafe/gfx/pggat/lib/fed"
type Matcher interface {
- Matches(conn fed.Conn) bool
+ Matches(conn *fed.Conn) bool
}
diff --git a/lib/gat/matchers/and.go b/lib/gat/matchers/and.go
index 7dd683b1..e4106ed9 100644
--- a/lib/gat/matchers/and.go
+++ b/lib/gat/matchers/and.go
@@ -45,7 +45,7 @@ func (T *And) Provision(ctx caddy.Context) error {
return nil
}
-func (T *And) Matches(conn fed.Conn) bool {
+func (T *And) Matches(conn *fed.Conn) bool {
for _, matcher := range T.and {
if !matcher.Matches(conn) {
return false
diff --git a/lib/gat/matchers/database.go b/lib/gat/matchers/database.go
index 0133a90e..d75731ac 100644
--- a/lib/gat/matchers/database.go
+++ b/lib/gat/matchers/database.go
@@ -24,8 +24,8 @@ func (T *Database) CaddyModule() caddy.ModuleInfo {
}
}
-func (T *Database) Matches(conn fed.Conn) bool {
- return conn.Database() == T.Database
+func (T *Database) Matches(conn *fed.Conn) bool {
+ return conn.Database == T.Database
}
var _ gat.Matcher = (*Database)(nil)
diff --git a/lib/gat/matchers/localaddress.go b/lib/gat/matchers/localaddress.go
index bd230eac..c9ac3093 100644
--- a/lib/gat/matchers/localaddress.go
+++ b/lib/gat/matchers/localaddress.go
@@ -47,8 +47,12 @@ func (T *LocalAddress) Provision(ctx caddy.Context) error {
return err
}
-func (T *LocalAddress) Matches(conn fed.Conn) bool {
- switch addr := conn.LocalAddr().(type) {
+func (T *LocalAddress) Matches(conn *fed.Conn) bool {
+ netConn, ok := conn.ReadWriteCloser.(*fed.NetConn)
+ if !ok {
+ return false
+ }
+ switch addr := netConn.LocalAddr().(type) {
case *net.TCPAddr:
expected, ok := T.addr.(*net.TCPAddr)
if !ok {
diff --git a/lib/gat/matchers/or.go b/lib/gat/matchers/or.go
index a8d0632c..baf7e89f 100644
--- a/lib/gat/matchers/or.go
+++ b/lib/gat/matchers/or.go
@@ -45,7 +45,7 @@ func (T *Or) Provision(ctx caddy.Context) error {
return nil
}
-func (T *Or) Matches(conn fed.Conn) bool {
+func (T *Or) Matches(conn *fed.Conn) bool {
for _, matcher := range T.or {
if matcher.Matches(conn) {
return true
diff --git a/lib/gat/matchers/ssl.go b/lib/gat/matchers/ssl.go
index eacd4090..57ae1a0a 100644
--- a/lib/gat/matchers/ssl.go
+++ b/lib/gat/matchers/ssl.go
@@ -24,8 +24,12 @@ func (T *SSL) CaddyModule() caddy.ModuleInfo {
}
}
-func (T *SSL) Matches(conn fed.Conn) bool {
- return conn.SSLEnabled() == T.SSL
+func (T *SSL) Matches(conn *fed.Conn) bool {
+ sslConn, ok := conn.ReadWriteCloser.(fed.SSL)
+ if !ok {
+ return T.SSL == false
+ }
+ return sslConn.SSL() == T.SSL
}
var _ gat.Matcher = (*SSL)(nil)
diff --git a/lib/gat/matchers/startupparameters.go b/lib/gat/matchers/startupparameters.go
index addb92f4..975e4d6e 100644
--- a/lib/gat/matchers/startupparameters.go
+++ b/lib/gat/matchers/startupparameters.go
@@ -36,10 +36,9 @@ func (T *StartupParameters) Provision(ctx caddy.Context) error {
return nil
}
-func (T *StartupParameters) Matches(conn fed.Conn) bool {
- initialParameters := conn.InitialParameters()
+func (T *StartupParameters) Matches(conn *fed.Conn) bool {
for key, value := range T.parameters {
- if initialParameters[key] != value {
+ if conn.InitialParameters[key] != value {
return false
}
}
diff --git a/lib/gat/matchers/user.go b/lib/gat/matchers/user.go
index a666a4bb..33f94cf0 100644
--- a/lib/gat/matchers/user.go
+++ b/lib/gat/matchers/user.go
@@ -24,8 +24,8 @@ func (T *User) CaddyModule() caddy.ModuleInfo {
}
}
-func (T *User) Matches(conn fed.Conn) bool {
- return conn.User() == T.User
+func (T *User) Matches(conn *fed.Conn) bool {
+ return conn.User == T.User
}
var _ gat.Matcher = (*User)(nil)
diff --git a/lib/gat/pool/client.go b/lib/gat/pool/client.go
index 72c6d8b6..c6d3c76e 100644
--- a/lib/gat/pool/client.go
+++ b/lib/gat/pool/client.go
@@ -2,11 +2,9 @@ package pool
import (
"gfx.cafe/gfx/pggat/lib/fed"
- "gfx.cafe/gfx/pggat/lib/middleware"
- "gfx.cafe/gfx/pggat/lib/middleware/interceptor"
- "gfx.cafe/gfx/pggat/lib/middleware/middlewares/eqp"
- "gfx.cafe/gfx/pggat/lib/middleware/middlewares/ps"
- "gfx.cafe/gfx/pggat/lib/middleware/middlewares/unterminate"
+ "gfx.cafe/gfx/pggat/lib/fed/middlewares/eqp"
+ "gfx.cafe/gfx/pggat/lib/fed/middlewares/ps"
+ "gfx.cafe/gfx/pggat/lib/fed/middlewares/unterminate"
)
type pooledClient struct {
@@ -18,31 +16,27 @@ type pooledClient struct {
func newClient(
options Config,
- conn fed.Conn,
+ conn *fed.Conn,
) *pooledClient {
- middlewares := []middleware.Middleware{
+ conn.Middleware = append(
+ conn.Middleware,
unterminate.Unterminate,
- }
+ )
var psClient *ps.Client
if options.ParameterStatusSync == ParameterStatusSyncDynamic {
// add ps middleware
- psClient = ps.NewClient(conn.InitialParameters())
- middlewares = append(middlewares, psClient)
+ psClient = ps.NewClient(conn.InitialParameters)
+ conn.Middleware = append(conn.Middleware, psClient)
}
var eqpClient *eqp.Client
if options.ExtendedQuerySync {
// add eqp middleware
eqpClient = eqp.NewClient()
- middlewares = append(middlewares, eqpClient)
+ conn.Middleware = append(conn.Middleware, eqpClient)
}
- conn = interceptor.NewInterceptor(
- conn,
- middlewares...,
- )
-
return &pooledClient{
pooledConn: makeConn(
conn,
diff --git a/lib/gat/pool/options.go b/lib/gat/pool/config.go
similarity index 100%
rename from lib/gat/pool/options.go
rename to lib/gat/pool/config.go
diff --git a/lib/gat/pool/conn.go b/lib/gat/pool/conn.go
index 07b54699..065a6458 100644
--- a/lib/gat/pool/conn.go
+++ b/lib/gat/pool/conn.go
@@ -15,9 +15,7 @@ import (
type pooledConn struct {
id uuid.UUID
- conn fed.Conn
- // please someone fix runtime.convI2I
- rw fed.ReadWriter
+ conn *fed.Conn
// metrics
@@ -35,12 +33,11 @@ type pooledConn struct {
}
func makeConn(
- conn fed.Conn,
+ conn *fed.Conn,
) pooledConn {
return pooledConn{
id: uuid.New(),
conn: conn,
- rw: conn,
since: time.Now(),
}
@@ -50,21 +47,16 @@ func (T *pooledConn) GetID() uuid.UUID {
return T.id
}
-func (T *pooledConn) GetConn() fed.Conn {
+func (T *pooledConn) GetConn() *fed.Conn {
return T.conn
}
-// GetReadWriter is the exact same as GetConn but bypasses the runtime.convI2I
-func (T *pooledConn) GetReadWriter() fed.ReadWriter {
- return T.rw
-}
-
func (T *pooledConn) GetInitialParameters() map[strutil.CIString]string {
- return T.conn.InitialParameters()
+ return T.conn.InitialParameters
}
func (T *pooledConn) GetBackendKey() [8]byte {
- return T.conn.BackendKey()
+ return T.conn.BackendKey
}
func (T *pooledConn) TransactionComplete() {
diff --git a/lib/gat/pool/flow.go b/lib/gat/pool/flow.go
index 72f0025e..e1263621 100644
--- a/lib/gat/pool/flow.go
+++ b/lib/gat/pool/flow.go
@@ -3,10 +3,10 @@ package pool
import (
"gfx.cafe/gfx/pggat/lib/bouncer/backends/v0"
"gfx.cafe/gfx/pggat/lib/fed"
+ "gfx.cafe/gfx/pggat/lib/fed/middlewares/eqp"
+ "gfx.cafe/gfx/pggat/lib/fed/middlewares/ps"
packets "gfx.cafe/gfx/pggat/lib/fed/packets/v3.0"
"gfx.cafe/gfx/pggat/lib/gat/metrics"
- "gfx.cafe/gfx/pggat/lib/middleware/middlewares/eqp"
- "gfx.cafe/gfx/pggat/lib/middleware/middlewares/ps"
"gfx.cafe/gfx/pggat/lib/util/slices"
)
@@ -23,7 +23,7 @@ func pair(options Config, client *pooledClient, server *pooledServer) (clientErr
switch options.ParameterStatusSync {
case ParameterStatusSyncDynamic:
- clientErr, serverErr = ps.Sync(options.TrackedParameters, client.GetReadWriter(), client.GetPS(), server.GetReadWriter(), server.GetPS())
+ clientErr, serverErr = ps.Sync(options.TrackedParameters, client.GetConn(), client.GetPS(), server.GetConn(), server.GetPS())
case ParameterStatusSyncInitial:
clientErr, serverErr = syncInitialParameters(options, client, server)
}
@@ -33,7 +33,7 @@ func pair(options Config, client *pooledClient, server *pooledServer) (clientErr
}
if options.ExtendedQuerySync {
- serverErr = eqp.Sync(client.GetEQP(), server.GetReadWriter(), server.GetEQP())
+ serverErr = eqp.Sync(client.GetEQP(), server.GetConn(), server.GetEQP())
}
return
@@ -80,7 +80,7 @@ func syncInitialParameters(options Config, client *pooledClient, server *pooledS
continue
}
- serverErr, _, packet = backends.SetParameter(server.GetReadWriter(), nil, packet, key, value)
+ serverErr, _, packet = backends.SetParameter(server.GetConn(), nil, packet, key, value)
if serverErr != nil {
return
}
diff --git a/lib/gat/pool/pool.go b/lib/gat/pool/pool.go
index dea718b7..91fc0ee3 100644
--- a/lib/gat/pool/pool.go
+++ b/lib/gat/pool/pool.go
@@ -267,7 +267,7 @@ func (T *Pool) releaseServer(server *pooledServer) {
if T.config.ServerResetQuery != "" {
server.SetState(metrics.ConnStateRunningResetQuery, uuid.Nil)
- err, _, _ := backends.QueryString(server.GetReadWriter(), nil, nil, T.config.ServerResetQuery)
+ err, _, _ := backends.QueryString(server.GetConn(), nil, nil, T.config.ServerResetQuery)
if err != nil {
T.removeServer(server)
return
@@ -280,7 +280,7 @@ func (T *Pool) releaseServer(server *pooledServer) {
}
func (T *Pool) Serve(
- conn fed.Conn,
+ conn *fed.Conn,
) error {
defer func() {
_ = conn.Close()
@@ -297,7 +297,7 @@ func (T *Pool) Serve(
// ServeBot is for clients that don't need initial parameters, cancelling queries, and are ready now. Use Serve for
// real clients
func (T *Pool) ServeBot(
- conn fed.Conn,
+ conn fed.ReadWriteCloser,
) error {
defer func() {
_ = conn.Close()
@@ -305,7 +305,9 @@ func (T *Pool) ServeBot(
client := newClient(
T.config,
- conn,
+ &fed.Conn{
+ ReadWriteCloser: conn,
+ },
)
return T.serve(client, true)
@@ -369,7 +371,7 @@ func (T *Pool) serve(client *pooledClient, initialized bool) error {
err, serverErr = pair(T.config, client, server)
}
if err == nil && serverErr == nil {
- packet, err, serverErr = bouncers.Bounce(client.GetReadWriter(), server.GetReadWriter(), packet)
+ packet, err, serverErr = bouncers.Bounce(client.GetConn(), server.GetConn(), packet)
}
if serverErr != nil {
return serverErr
diff --git a/lib/gat/pool/recipe/options.go b/lib/gat/pool/recipe/config.go
similarity index 100%
rename from lib/gat/pool/recipe/options.go
rename to lib/gat/pool/recipe/config.go
diff --git a/lib/gat/pool/recipe/dialer.go b/lib/gat/pool/recipe/dialer.go
index d40ad740..e9392c11 100644
--- a/lib/gat/pool/recipe/dialer.go
+++ b/lib/gat/pool/recipe/dialer.go
@@ -25,16 +25,18 @@ type Dialer struct {
StartupParameters map[strutil.CIString]string
}
-func (T Dialer) Dial() (fed.Conn, error) {
+func (T Dialer) Dial() (*fed.Conn, error) {
c, err := net.Dial(T.Network, T.Address)
if err != nil {
return nil, err
}
- conn := fed.WrapNetConn(c)
- conn.SetUser(T.Username)
- conn.SetDatabase(T.Database)
- _, initialParameters, backendKey, err := backends.Accept(
- conn,
+ conn := fed.NewConn(
+ fed.NewNetConn(c),
+ )
+ conn.User = T.Username
+ conn.Database = T.Database
+ _, conn.InitialParameters, conn.BackendKey, err = backends.Accept(
+ conn.ReadWriteCloser,
T.SSLMode,
T.SSLConfig,
T.Username,
@@ -45,8 +47,6 @@ func (T Dialer) Dial() (fed.Conn, error) {
if err != nil {
return nil, err
}
- conn.SetInitialParameters(initialParameters)
- conn.SetBackendKey(backendKey)
return conn, nil
}
@@ -55,7 +55,7 @@ func (T Dialer) Cancel(key [8]byte) error {
if err != nil {
return err
}
- conn := fed.WrapNetConn(c)
+ conn := fed.NewNetConn(c)
defer func() {
_ = conn.Close()
}()
diff --git a/lib/gat/pool/recipe/recipe.go b/lib/gat/pool/recipe/recipe.go
index d4e7165c..ba02bd09 100644
--- a/lib/gat/pool/recipe/recipe.go
+++ b/lib/gat/pool/recipe/recipe.go
@@ -66,7 +66,7 @@ func (T *Recipe) Free() {
T.count--
}
-func (T *Recipe) Dial() (fed.Conn, error) {
+func (T *Recipe) Dial() (*fed.Conn, error) {
return T.config.Dialer.Dial()
}
diff --git a/lib/gat/pool/server.go b/lib/gat/pool/server.go
index 39213244..9afa7c99 100644
--- a/lib/gat/pool/server.go
+++ b/lib/gat/pool/server.go
@@ -2,10 +2,8 @@ package pool
import (
"gfx.cafe/gfx/pggat/lib/fed"
- "gfx.cafe/gfx/pggat/lib/middleware"
- "gfx.cafe/gfx/pggat/lib/middleware/interceptor"
- "gfx.cafe/gfx/pggat/lib/middleware/middlewares/eqp"
- "gfx.cafe/gfx/pggat/lib/middleware/middlewares/ps"
+ "gfx.cafe/gfx/pggat/lib/fed/middlewares/eqp"
+ "gfx.cafe/gfx/pggat/lib/fed/middlewares/ps"
)
type pooledServer struct {
@@ -20,29 +18,20 @@ type pooledServer struct {
func newServer(
options Config,
recipe string,
- conn fed.Conn,
+ conn *fed.Conn,
) *pooledServer {
- var middlewares []middleware.Middleware
-
var psServer *ps.Server
if options.ParameterStatusSync == ParameterStatusSyncDynamic {
// add ps middleware
- psServer = ps.NewServer(conn.InitialParameters())
- middlewares = append(middlewares, psServer)
+ psServer = ps.NewServer(conn.InitialParameters)
+ conn.Middleware = append(conn.Middleware, psServer)
}
var eqpServer *eqp.Server
if options.ExtendedQuerySync {
// add eqp middleware
eqpServer = eqp.NewServer()
- middlewares = append(middlewares, eqpServer)
- }
-
- if len(middlewares) > 0 {
- conn = interceptor.NewInterceptor(
- conn,
- middlewares...,
- )
+ conn.Middleware = append(conn.Middleware, eqpServer)
}
return &pooledServer{
diff --git a/lib/gat/provider.go b/lib/gat/provider.go
index 127e16a7..a3162f98 100644
--- a/lib/gat/provider.go
+++ b/lib/gat/provider.go
@@ -7,6 +7,6 @@ import (
// Provider provides pool to the server
type Provider interface {
- Lookup(conn fed.Conn) *Pool
+ Lookup(conn *fed.Conn) *Pool
ReadMetrics(metrics *metrics.Pools)
}
diff --git a/lib/gat/providers/discovery/discoverers/google_cloud_sql/discoverer.go b/lib/gat/providers/discovery/discoverers/google_cloud_sql/discoverer.go
index b6b66538..41ba08e4 100644
--- a/lib/gat/providers/discovery/discoverers/google_cloud_sql/discoverer.go
+++ b/lib/gat/providers/discovery/discoverers/google_cloud_sql/discoverer.go
@@ -97,7 +97,7 @@ func (T *Discoverer) instanceToCluster(primary *sqladmin.DatabaseInstance, repli
return c, nil
}
- var admin fed.Conn
+ var admin *fed.Conn
defer func() {
if admin != nil {
_ = admin.Close()
diff --git a/lib/gat/providers/discovery/module.go b/lib/gat/providers/discovery/module.go
index 07fe74da..edeb339f 100644
--- a/lib/gat/providers/discovery/module.go
+++ b/lib/gat/providers/discovery/module.go
@@ -524,8 +524,8 @@ func (T *Module) lookup(user, database string) *gat.Pool {
return p
}
-func (T *Module) Lookup(conn fed.Conn) *gat.Pool {
- return T.lookup(conn.User(), conn.Database())
+func (T *Module) Lookup(conn *fed.Conn) *gat.Pool {
+ return T.lookup(conn.User, conn.Database)
}
var _ gat.Provider = (*Module)(nil)
diff --git a/lib/gat/providers/pgbouncer/module.go b/lib/gat/providers/pgbouncer/module.go
index b3bdc582..a5705d76 100644
--- a/lib/gat/providers/pgbouncer/module.go
+++ b/lib/gat/providers/pgbouncer/module.go
@@ -273,8 +273,8 @@ func (T *Module) lookup(user, database string) *gat.Pool {
return T.tryCreate(user, database)
}
-func (T *Module) Lookup(conn fed.Conn) *gat.Pool {
- return T.lookup(conn.User(), conn.Database())
+func (T *Module) Lookup(conn *fed.Conn) *gat.Pool {
+ return T.lookup(conn.User, conn.Database)
}
func (T *Module) ReadMetrics(metrics *metrics.Pools) {
diff --git a/lib/gat/server.go b/lib/gat/server.go
index 23a01606..04645592 100644
--- a/lib/gat/server.go
+++ b/lib/gat/server.go
@@ -47,7 +47,7 @@ func (T *Server) Provision(ctx caddy.Context) error {
return nil
}
-func (T *Server) lookup(conn fed.Conn) *pool.Pool {
+func (T *Server) lookup(conn *fed.Conn) *pool.Pool {
for _, route := range T.routes {
if route.match != nil && !route.match.Matches(conn) {
continue
diff --git a/lib/gsql/client.go b/lib/gsql/client.go
index f7adaccf..de0a0630 100644
--- a/lib/gsql/client.go
+++ b/lib/gsql/client.go
@@ -8,7 +8,6 @@ import (
"gfx.cafe/gfx/pggat/lib/fed"
"gfx.cafe/gfx/pggat/lib/util/ring"
"gfx.cafe/gfx/pggat/lib/util/slices"
- "gfx.cafe/gfx/pggat/lib/util/strutil"
)
type batch struct {
@@ -144,32 +143,4 @@ func (T *Client) Close() error {
return nil
}
-func (T *Client) LocalAddr() net.Addr {
- return Addr{}
-}
-
-func (T *Client) RemoteAddr() net.Addr {
- return Addr{}
-}
-
-func (T *Client) SSLEnabled() bool {
- return false
-}
-
-func (T *Client) User() string {
- return ""
-}
-
-func (T *Client) Database() string {
- return ""
-}
-
-func (T *Client) InitialParameters() map[strutil.CIString]string {
- return nil
-}
-
-func (T *Client) BackendKey() [8]byte {
- return [8]byte{}
-}
-
-var _ fed.Conn = (*Client)(nil)
+var _ fed.ReadWriteCloser = (*Client)(nil)
diff --git a/lib/middleware/context.go b/lib/middleware/context.go
deleted file mode 100644
index be60aa0e..00000000
--- a/lib/middleware/context.go
+++ /dev/null
@@ -1,11 +0,0 @@
-package middleware
-
-import "gfx.cafe/gfx/pggat/lib/fed"
-
-type Context interface {
- // Cancel the current packet
- Cancel()
-
- // Write packet to underlying connection
- Write(packet fed.Packet) error
-}
diff --git a/lib/middleware/interceptor/context.go b/lib/middleware/interceptor/context.go
deleted file mode 100644
index ea4c7a76..00000000
--- a/lib/middleware/interceptor/context.go
+++ /dev/null
@@ -1,36 +0,0 @@
-package interceptor
-
-import (
- "gfx.cafe/gfx/pggat/lib/fed"
- "gfx.cafe/gfx/pggat/lib/middleware"
- "gfx.cafe/gfx/pggat/lib/util/decorator"
-)
-
-type Context struct {
- noCopy decorator.NoCopy
-
- cancelled bool
-
- // for normal Write / WriteUntyped
- rw fed.ReadWriter
-}
-
-func makeContext(rw fed.ReadWriter) Context {
- return Context{
- rw: rw,
- }
-}
-
-func (T *Context) reset() {
- T.cancelled = false
-}
-
-func (T *Context) Cancel() {
- T.cancelled = true
-}
-
-func (T *Context) Write(packet fed.Packet) error {
- return T.rw.WritePacket(packet)
-}
-
-var _ middleware.Context = (*Context)(nil)
diff --git a/lib/middleware/interceptor/interceptor.go b/lib/middleware/interceptor/interceptor.go
deleted file mode 100644
index 7336962a..00000000
--- a/lib/middleware/interceptor/interceptor.go
+++ /dev/null
@@ -1,100 +0,0 @@
-package interceptor
-
-import (
- "net"
-
- "gfx.cafe/gfx/pggat/lib/fed"
- "gfx.cafe/gfx/pggat/lib/middleware"
- "gfx.cafe/gfx/pggat/lib/util/strutil"
-)
-
-type Interceptor struct {
- middlewares []middleware.Middleware
- context Context
- conn fed.Conn
-}
-
-func NewInterceptor(conn fed.Conn, middlewares ...middleware.Middleware) *Interceptor {
- if v, ok := conn.(*Interceptor); ok {
- v.middlewares = append(v.middlewares, middlewares...)
- return v
- }
- return &Interceptor{
- middlewares: middlewares,
- context: makeContext(conn),
- conn: conn,
- }
-}
-
-func (T *Interceptor) ReadPacket(typed bool, packet fed.Packet) (fed.Packet, error) {
-outer:
- for {
- var err error
- packet, err = T.conn.ReadPacket(typed, packet)
- if err != nil {
- return packet, err
- }
-
- for _, mw := range T.middlewares {
- T.context.reset()
- err = mw.Read(&T.context, packet)
- if err != nil {
- return packet, err
- }
- if T.context.cancelled {
- continue outer
- }
- }
-
- return packet, nil
- }
-}
-
-func (T *Interceptor) WritePacket(packet fed.Packet) error {
- for _, mw := range T.middlewares {
- T.context.reset()
- err := mw.Write(&T.context, packet)
- if err != nil {
- return err
- }
- if T.context.cancelled {
- return nil
- }
- }
-
- return T.conn.WritePacket(packet)
-}
-
-func (T *Interceptor) LocalAddr() net.Addr {
- return T.conn.LocalAddr()
-}
-
-func (T *Interceptor) RemoteAddr() net.Addr {
- return T.conn.RemoteAddr()
-}
-
-func (T *Interceptor) SSLEnabled() bool {
- return T.conn.SSLEnabled()
-}
-
-func (T *Interceptor) User() string {
- return T.conn.User()
-}
-
-func (T *Interceptor) Database() string {
- return T.conn.Database()
-}
-
-func (T *Interceptor) InitialParameters() map[strutil.CIString]string {
- return T.conn.InitialParameters()
-}
-
-func (T *Interceptor) BackendKey() [8]byte {
- return T.conn.BackendKey()
-}
-
-func (T *Interceptor) Close() error {
- return T.conn.Close()
-}
-
-var _ fed.Conn = (*Interceptor)(nil)
diff --git a/lib/middleware/middleware.go b/lib/middleware/middleware.go
deleted file mode 100644
index 63f5401d..00000000
--- a/lib/middleware/middleware.go
+++ /dev/null
@@ -1,8 +0,0 @@
-package middleware
-
-import "gfx.cafe/gfx/pggat/lib/fed"
-
-type Middleware interface {
- Read(ctx Context, packet fed.Packet) error
- Write(ctx Context, packet fed.Packet) error
-}
diff --git a/lib/middleware/middlewares/eqp/client.go b/lib/middleware/middlewares/eqp/client.go
deleted file mode 100644
index 89175b71..00000000
--- a/lib/middleware/middlewares/eqp/client.go
+++ /dev/null
@@ -1,26 +0,0 @@
-package eqp
-
-import (
- "gfx.cafe/gfx/pggat/lib/fed"
- "gfx.cafe/gfx/pggat/lib/middleware"
-)
-
-type Client struct {
- state State
-}
-
-func NewClient() *Client {
- return new(Client)
-}
-
-func (T *Client) Read(_ middleware.Context, packet fed.Packet) error {
- T.state.C2S(packet)
- return nil
-}
-
-func (T *Client) Write(_ middleware.Context, packet fed.Packet) error {
- T.state.S2C(packet)
- return nil
-}
-
-var _ middleware.Middleware = (*Client)(nil)
diff --git a/lib/middleware/middlewares/eqp/server.go b/lib/middleware/middlewares/eqp/server.go
deleted file mode 100644
index 04a11d17..00000000
--- a/lib/middleware/middlewares/eqp/server.go
+++ /dev/null
@@ -1,26 +0,0 @@
-package eqp
-
-import (
- "gfx.cafe/gfx/pggat/lib/fed"
- "gfx.cafe/gfx/pggat/lib/middleware"
-)
-
-type Server struct {
- state State
-}
-
-func NewServer() *Server {
- return new(Server)
-}
-
-func (T *Server) Read(_ middleware.Context, packet fed.Packet) error {
- T.state.S2C(packet)
- return nil
-}
-
-func (T *Server) Write(_ middleware.Context, packet fed.Packet) error {
- T.state.C2S(packet)
- return nil
-}
-
-var _ middleware.Middleware = (*Server)(nil)
--
GitLab