First steps towards implementing the PostgreSQL protocol.

server/testserver.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/gossip"
 	"github.com/cockroachdb/cockroach/proto"
 	"github.com/cockroachdb/cockroach/security"
+	"github.com/cockroachdb/cockroach/sql"
 	"github.com/cockroachdb/cockroach/storage"
 	"github.com/cockroachdb/cockroach/storage/engine"
 	"github.com/cockroachdb/cockroach/ts"
@@ -131,6 +132,11 @@ func (ts *TestServer) EventFeed() *util.Feed {
 	return nil
 }
 
+// SQLExecutor returns the sql.Executor used by the TestServer.
+func (ts *TestServer) SQLExecutor() sql.Executor {
+	return ts.Server.sqlServer.Executor
+}
+
 // Start starts the TestServer by bootstrapping an in-memory store
 // (defaults to maximum of 100M). The server is started, launching the
 // node RPC server and all HTTP endpoints. Use the value of

sql/pgwire/context.go

@@ -0,0 +1,31 @@
+// Copyright 2015 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License. See the AUTHORS file
+// for names of contributors.
+//
+// Author: Ben Darnell
+
+package pgwire
+
+import (
+	"github.com/cockroachdb/cockroach/base"
+	"github.com/cockroachdb/cockroach/sql"
+	"github.com/cockroachdb/cockroach/util/stop"
+)
+
+// Context holds parameters needed to setup a postgres-compatible server.
+type Context struct {
+	*base.Context
+	Executor sql.Executor
+	Stopper  *stop.Stopper
+}

sql/pgwire/encoding.go

@@ -0,0 +1,98 @@
+// Copyright 2015 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License. See the AUTHORS file
+// for names of contributors.
+//
+// Author: Ben Darnell
+
+package pgwire
+
+import (
+	"encoding/binary"
+	"io"
+
+	"github.com/cockroachdb/cockroach/util"
+)
+
+const maxMessageSize = 1 << 24
+
+// bufferedReader is implemented by bufio.Reader and bytes.Buffer.
+type bufferedReader interface {
+	io.Reader
+	ReadString(delim byte) (string, error)
+	ReadByte() (byte, error)
+}
+
+// readMsg reads a length-prefixed message. It is only used directly
+// during the authentication phase of the protocol; readTypedMsg is
+// used at all other times.
+func readUntypedMsg(rd io.Reader) ([]byte, error) {
+	var size int32
+	if err := binary.Read(rd, binary.BigEndian, &size); err != nil {
+		return nil, err
+	}
+	// size includes itself.
+	size -= 4
+	if size > maxMessageSize || size < 0 {
+		return nil, util.Errorf("message size %d out of bounds (0..%d)",
+			size, maxMessageSize)
+	}
+	data := make([]byte, size)
+	if _, err := io.ReadFull(rd, data); err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
+// readTypedMsg reads a message, returning its type code and body.
+func readTypedMsg(rd bufferedReader) (messageType, []byte, error) {
+	typ, err := rd.ReadByte()
+	if err != nil {
+		return 0, nil, err
+	}
+	msg, err := readUntypedMsg(rd)
+	return messageType(typ), msg, err
+}
+
+// writeTypedMsg writes a message with the given type.
+func writeTypedMsg(w io.Writer, typ messageType, msg []byte) error {
+	if _, err := w.Write([]byte{byte(typ)}); err != nil {
+		return err
+	}
+	if err := binary.Write(w, binary.BigEndian, int32(len(msg)+4)); err != nil {
+		return nil
+	}
+	_, err := w.Write(msg)
+	return err
+}
+
+// readString reads a null-terminated string.
+func readString(rd bufferedReader) (string, error) {
+	s, err := rd.ReadString(0)
+	if err != nil {
+		return "", err
+	}
+	// Chop off the trailing delimiter.
+	return s[:len(s)-1], nil
+}
+
+// writeString writes a null-terminated string.
+func writeString(w io.Writer, s string) error {
+	if _, err := io.WriteString(w, s); err != nil {
+		return err
+	}
+	if _, err := w.Write([]byte{0}); err != nil {
+		return err
+	}
+	return nil
+}

sql/pgwire/main_test.go

@@ -0,0 +1,36 @@
+// Copyright 2015 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License. See the AUTHORS file
+// for names of contributors.
+//
+// Author: Ben Darnell
+
+package pgwire
+
+import (
+	"testing"
+
+	"github.com/cockroachdb/cockroach/security"
+	"github.com/cockroachdb/cockroach/security/securitytest"
+	"github.com/cockroachdb/cockroach/util/leaktest"
+)
+
+func init() {
+	security.SetReadFileFn(securitytest.Asset)
+}
+
+//go:generate ../../util/leaktest/add-leaktest.sh *_test.go
+
+func TestMain(m *testing.M) {
+	leaktest.TestMainWithLeakCheck(m)
+}

sql/pgwire/server.go

@@ -0,0 +1,122 @@
+// Copyright 2015 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License. See the AUTHORS file
+// for names of contributors.
+//
+// Author: Ben Darnell
+
+package pgwire
+
+import (
+	"bytes"
+	"net"
+
+	"github.com/cockroachdb/cockroach/util"
+	"github.com/cockroachdb/cockroach/util/log"
+)
+
+var (
+	versionSSL = []byte{0x12, 0x34, 0x56, 0x79}
+	version30  = []byte{0x00, 0x03, 0x00, 0x00}
+)
+
+// Server implements the server side of the PostgreSQL wire protocol.
+type Server struct {
+	context  *Context
+	listener net.Listener
+	conns    []net.Conn
+}
+
+// NewServer creates a Server.
+func NewServer(context *Context) *Server {
+	s := &Server{
+		context: context,
+	}
+	return s
+}
+
+// Start a server on the given address.
+func (s *Server) Start(addr net.Addr) error {
+	ln, err := net.Listen(addr.Network(), addr.String())
+	if err != nil {
+		return err
+	}
+	s.listener = ln
+
+	s.context.Stopper.RunWorker(func() {
+		s.serve(ln)
+	})
+
+	s.context.Stopper.RunWorker(func() {
+		<-s.context.Stopper.ShouldStop()
+		s.close()
+	})
+	return nil
+}
+
+// Addr returns this Server's address.
+func (s *Server) Addr() net.Addr {
+	return s.listener.Addr()
+}
+
+// serve connections on this listener until it is closed.
+func (s *Server) serve(ln net.Listener) {
+	for {
+		conn, err := ln.Accept()
+		if err != nil {
+			// TODO(bdarnell): error handling
+			log.Error(err)
+			return
+		}
+
+		s.conns = append(s.conns, conn)
+		go func() {
+			if err := s.serveConn(conn); err != nil {
+				log.Error(err)
+			}
+		}()
+	}
+}
+
+// close this server, and all client connections.
+func (s *Server) close() {
+	s.listener.Close()
+	for _, conn := range s.conns {
+		conn.Close()
+	}
+}
+
+// serveConn serves a single connection, driving the handshake process
+// and delegating to the appropriate connection type.
+func (s *Server) serveConn(conn net.Conn) error {
+	defer conn.Close()
+	msg, err := readUntypedMsg(conn)
+	if err != nil {
+		return err
+	}
+	version := msg[:4]
+	rest := msg[4:]
+	if bytes.Compare(version, versionSSL) == 0 {
+		if len(rest) > 0 {
+			return util.Errorf("unexpected data after SSL request")
+		}
+		panic("TODO(bdarnell): ssl mode")
+	} else if bytes.Compare(version, version30) == 0 {
+		v3conn, err := newV3Conn(conn, rest, s.context.Executor)
+		if err != nil {
+			return err
+		}
+		return v3conn.serve()
+	}
+	return util.Errorf("unknown protocol version %q", version)
+}