Merge remote-tracking branch 'origin/master'

# Conflicts:
#	protocol/transport_test.go

protocol/extend.go

@@ -0,0 +1,51 @@
+package protocol
+
+// ParseComplete is sent when backend parsed a prepared statement successfully
+var ParseComplete = []byte{'1', 0, 0, 0, 4}
+
+// BindComplete is sent when backend prepared a portal and finished planning the query
+var BindComplete = []byte{'2', 0, 0, 0, 4}
+
+// CreatesTransaction tells weather this is a frontend message that should start/continue a transaction
+func (m *Message) CreatesTransaction() bool {
+	return m.Type() == Parse || m.Type() == Bind
+}
+
+// EndsTransaction tells weather this is a frontend message that should end the current transaction
+func (m *Message) EndsTransaction() bool {
+	return m.Type() == Query || m.Type() == Sync
+}
+
+// transaction represents a sequence of frontend and backend messages
+// that apply only on commit. the purpose of transaction is to support
+// extended query flow.
+type transaction struct {
+	transport *Transport
+	in        []Message // TODO: asses if we need it after implementation of prepared statements and portals is done
+	out       []Message // TODO: add size limit
+}
+
+// Read uses Transport to read the next message into the transaction's incoming messages buffer
+func (t *transaction) Read() (msg Message, err error) {
+	if msg, err = t.transport.read(); err == nil {
+		t.in = append(t.in, msg)
+	}
+	return
+}
+
+// Write writes the provided message into the transaction's outgoing messages buffer
+func (t *transaction) Write(msg Message) error {
+	t.out = append(t.out, msg)
+	return nil
+}
+
+func (t *transaction) flush() (err error) {
+	for len(t.out) > 0 {
+		err = t.transport.write(t.out[0])
+		if err != nil {
+			break
+		}
+		t.out = t.out[1:]
+	}
+	return
+}

protocol/transaction_test.go → protocol/extend_test.go

@@ -8,11 +8,10 @@ import (
 )
 
 func TestTransaction_Read(t *testing.T) {
-
 	buf := bytes.Buffer{}
 	comm := bufio.NewReadWriter(bufio.NewReader(&buf), bufio.NewWriter(&buf))
-	p := &Protocol{W: comm, R: comm, initialized: true}
-	trans := &transaction{p: p, in: []Message{}, out: []Message{}}
+	p := &Transport{W: comm, R: comm, initialized: true}
+	trans := &transaction{transport: p, in: []Message{}, out: []Message{}}
 
 	_, err := comm.Write([]byte{'P', 0, 0, 0, 4})
 	require.NoError(t, err)
@@ -22,8 +21,8 @@ func TestTransaction_Read(t *testing.T) {
 
 	m, err := trans.Read()
 	require.NoError(t, err)
-	require.NotNilf(t, m,
-		"expected exactly 1 message in transaction incoming buffer. actual: %d", len(trans.in))
+	require.NotNil(t, m,
+		"expected to receive message from transaction. got nil")
 
 	require.Equalf(t, 1, len(trans.in),
 		"expected exactly 1 message in transaction incoming buffer. actual: %d", len(trans.in))
@@ -39,5 +38,4 @@ func TestTransaction_Read(t *testing.T) {
 
 	require.Equalf(t, 1, len(trans.out),
 		"expected exactly one message in transaction's outgoind message buffer. actual messages count: %d", len(trans.out))
-
 }

protocol/message.go

@@ -28,8 +28,8 @@ func (m Message) Type() byte {
 }
 
 // MessageReadWriter describes objects that handle client-server communication.
-// Objects implementing this interface are used to send password requests to users,
-// and receive their responses.
+// Objects implementing this interface are used by logic operations to send Message
+// objects to frontend and receive Message back from it
 type MessageReadWriter interface {
 	Write(m Message) error
 	Read() (Message, error)