ccl/sqlproxyccl: cleanup PG interceptor APIs

pkg/ccl/sqlproxyccl/interceptor/backend_interceptor.go

@@ -18,14 +18,10 @@ import (
 // BackendInterceptor is a server int/erceptor for the Postgres backend protocol.
 type BackendInterceptor pgInterceptor
 
-// NewBackendInterceptor creates a BackendInterceptor. bufSize must be at least
-// the size of a pgwire message header.
-func NewBackendInterceptor(src io.Reader, dst io.Writer, bufSize int) (*BackendInterceptor, error) {
-	pgi, err := newPgInterceptor(src, dst, bufSize)
-	if err != nil {
-		return nil, err
-	}
-	return (*BackendInterceptor)(pgi), nil
+// NewBackendInterceptor creates a BackendInterceptor using the default buffer
+// size of 8K bytes.
+func NewBackendInterceptor(src io.Reader) *BackendInterceptor {
+	return (*BackendInterceptor)(newPgInterceptor(src, defaultBufferSize))
 }
 
 // PeekMsg returns the header of the current pgwire message without advancing
@@ -37,13 +33,6 @@ func (bi *BackendInterceptor) PeekMsg() (typ pgwirebase.ClientMessageType, size
 	return pgwirebase.ClientMessageType(byteType), size, err
 }
 
-// WriteMsg writes the given bytes to the writer dst.
-//
-// See pgInterceptor.WriteMsg for more information.
-func (bi *BackendInterceptor) WriteMsg(data pgproto3.FrontendMessage) (n int, err error) {
-	return (*pgInterceptor)(bi).WriteMsg(data.Encode(nil))
-}
-
 // ReadMsg decodes the current pgwire message and returns a FrontendMessage.
 // This also advances the interceptor to the next message.
 //
@@ -53,19 +42,14 @@ func (bi *BackendInterceptor) ReadMsg() (msg pgproto3.FrontendMessage, err error
 	if err != nil {
 		return nil, err
 	}
-	// errPanicWriter is used here because Receive must not Write.
-	return pgproto3.NewBackend(newChunkReader(msgBytes), &errPanicWriter{}).Receive()
+	// errWriter is used here because Receive must not Write.
+	return pgproto3.NewBackend(newChunkReader(msgBytes), &errWriter{}).Receive()
 }
 
 // ForwardMsg sends the current pgwire message to the destination without any
 // decoding, and advances the interceptor to the next message.
 //
 // See pgInterceptor.ForwardMsg for more information.
-func (bi *BackendInterceptor) ForwardMsg() (n int, err error) {
-	return (*pgInterceptor)(bi).ForwardMsg()
-}
-
-// Close closes the interceptor, and prevents further operations on it.
-func (bi *BackendInterceptor) Close() {
-	(*pgInterceptor)(bi).Close()
+func (bi *BackendInterceptor) ForwardMsg(dst io.Writer) (n int, err error) {
+	return (*pgInterceptor)(bi).ForwardMsg(dst)
 }

pkg/ccl/sqlproxyccl/interceptor/backend_interceptor_test.go

@@ -27,91 +27,49 @@ func TestBackendInterceptor(t *testing.T) {
 
 	q := (&pgproto3.Query{String: "SELECT 1"}).Encode(nil)
 
-	t.Run("bufSize too small", func(t *testing.T) {
-		bi, err := interceptor.NewBackendInterceptor(nil /* src */, nil /* dst */, 1)
-		require.Error(t, err)
-		require.Nil(t, bi)
-	})
-
-	t.Run("PeekMsg returns the right message type", func(t *testing.T) {
+	buildSrc := func(t *testing.T, count int) *bytes.Buffer {
+		t.Helper()
 		src := new(bytes.Buffer)
 		_, err := src.Write(q)
 		require.NoError(t, err)
+		return src
+	}
 
-		bi, err := interceptor.NewBackendInterceptor(src, nil /* dst */, 16)
-		require.NoError(t, err)
+	t.Run("PeekMsg returns the right message type", func(t *testing.T) {
+		src := buildSrc(t, 1)
+
+		bi := interceptor.NewBackendInterceptor(src)
 		require.NotNil(t, bi)
 
 		typ, size, err := bi.PeekMsg()
 		require.NoError(t, err)
 		require.Equal(t, pgwirebase.ClientMsgSimpleQuery, typ)
 		require.Equal(t, 14, size)
-
-		bi.Close()
-		typ, size, err = bi.PeekMsg()
-		require.EqualError(t, err, interceptor.ErrInterceptorClosed.Error())
-		require.Equal(t, pgwirebase.ClientMessageType(0), typ)
-		require.Equal(t, 0, size)
-	})
-
-	t.Run("WriteMsg writes data to dst", func(t *testing.T) {
-		dst := new(bytes.Buffer)
-		bi, err := interceptor.NewBackendInterceptor(nil /* src */, dst, 10)
-		require.NoError(t, err)
-		require.NotNil(t, bi)
-
-		// This is a backend interceptor, so writing goes to the server.
-		toSend := &pgproto3.Query{String: "SELECT 1"}
-		n, err := bi.WriteMsg(toSend)
-		require.NoError(t, err)
-		require.Equal(t, 14, n)
-		require.Equal(t, 14, dst.Len())
-
-		bi.Close()
-		n, err = bi.WriteMsg(toSend)
-		require.EqualError(t, err, interceptor.ErrInterceptorClosed.Error())
-		require.Equal(t, 0, n)
 	})
 
 	t.Run("ReadMsg decodes the message correctly", func(t *testing.T) {
-		src := new(bytes.Buffer)
-		_, err := src.Write(q)
-		require.NoError(t, err)
+		src := buildSrc(t, 1)
 
-		bi, err := interceptor.NewBackendInterceptor(src, nil /* dst */, 16)
-		require.NoError(t, err)
+		bi := interceptor.NewBackendInterceptor(src)
 		require.NotNil(t, bi)
 
 		msg, err := bi.ReadMsg()
 		require.NoError(t, err)
 		rmsg, ok := msg.(*pgproto3.Query)
 		require.True(t, ok)
 		require.Equal(t, "SELECT 1", rmsg.String)
-
-		bi.Close()
-		msg, err = bi.ReadMsg()
-		require.EqualError(t, err, interceptor.ErrInterceptorClosed.Error())
-		require.Nil(t, msg)
 	})
 
 	t.Run("ForwardMsg forwards data to dst", func(t *testing.T) {
-		src := new(bytes.Buffer)
-		_, err := src.Write(q)
-		require.NoError(t, err)
+		src := buildSrc(t, 1)
 		dst := new(bytes.Buffer)
 
-		bi, err := interceptor.NewBackendInterceptor(src, dst, 16)
-		require.NoError(t, err)
+		bi := interceptor.NewBackendInterceptor(src)
 		require.NotNil(t, bi)
 
-		n, err := bi.ForwardMsg()
+		n, err := bi.ForwardMsg(dst)
 		require.NoError(t, err)
 		require.Equal(t, 14, n)
 		require.Equal(t, 14, dst.Len())
-
-		bi.Close()
-		n, err = bi.ForwardMsg()
-		require.EqualError(t, err, interceptor.ErrInterceptorClosed.Error())
-		require.Equal(t, 0, n)
 	})
 }