streams: interactive transactions and support

The main purpose of streams is transactions via iproto.
Since v. 2.10.0, Tarantool supports streams and
interactive transactions over them. Each stream
can start its own transaction, so they allows
multiplexing several transactions over one connection.

API for this feature is the following:

* `NewStream()` method to create a stream object for `Connection`
   and `NewStream(userMode Mode)` method to create a stream object
   for `ConnectionPool`
*  stream object `Stream` with `Do()`, `Begin()`,
   `Commit()`, `Rollback()` methods,
   `Begin()` - start transaction via iproto stream;
   `Commit()` - commit transaction;
   `Rollback()` - rollback transaction.

Closes #101

Author: AnaNek

CHANGELOG.md

@@ -15,6 +15,7 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 - Public API with request object types (#126)
 - Support decimal type in msgpack (#96)
 - Support datetime type in msgpack (#118)
+- Streams and interactive transactions support (#101)
 
 ### Changed
 

connection.go

@@ -19,6 +19,7 @@ import (
 )
 
 const requestsMap = 128
+const ignoreStreamId = 0
 const (
 	connDisconnected = 0
 	connConnected    = 1
@@ -139,6 +140,8 @@ type Connection struct {
 	state   uint32
 	dec     *msgpack.Decoder
 	lenbuf  [PacketLengthBytes]byte
+
+	lastStreamId uint32
 }
 
 var _ = Connector(&Connection{}) // Check compatibility with connector interface.
@@ -468,16 +471,27 @@ func (conn *Connection) dial() (err error) {
 	return
 }
 
-func pack(h *smallWBuf, enc *msgpack.Encoder, reqid uint32, req Request, res SchemaResolver) (err error) {
+func pack(h *smallWBuf, enc *msgpack.Encoder, reqid uint32,
+	req Request, streamId uint32, res SchemaResolver) (err error) {
 	hl := h.Len()
-	h.Write([]byte{
+
+	hMapLen := byte(0x82) // 2 element map.
+	if streamId != ignoreStreamId {
+		hMapLen = byte(0x83) // 3 element map.
+	}
+	hBytes := []byte{
 		0xce, 0, 0, 0, 0, // Length.
-		0x82,                      // 2 element map.
+		hMapLen,
 		KeyCode, byte(req.Code()), // Request code.
 		KeySync, 0xce,
 		byte(reqid >> 24), byte(reqid >> 16),
 		byte(reqid >> 8), byte(reqid),
-	})
+	}
+	if streamId != ignoreStreamId {
+		hBytes = append(hBytes, KeyStreamId, byte(streamId))
+	}
+
+	h.Write(hBytes)
 
 	if err = req.Body(res, enc); err != nil {
 		return
@@ -495,7 +509,7 @@ func pack(h *smallWBuf, enc *msgpack.Encoder, reqid uint32, req Request, res Sch
 func (conn *Connection) writeAuthRequest(w *bufio.Writer, scramble []byte) (err error) {
 	var packet smallWBuf
 	req := newAuthRequest(conn.opts.User, string(scramble))
-	err = pack(&packet, msgpack.NewEncoder(&packet), 0, req, conn.Schema)
+	err = pack(&packet, msgpack.NewEncoder(&packet), 0, req, ignoreStreamId, conn.Schema)
 
 	if err != nil {
 		return errors.New("auth: pack error " + err.Error())
@@ -785,16 +799,16 @@ func (conn *Connection) newFuture() (fut *Future) {
 	return
 }
 
-func (conn *Connection) send(req Request) *Future {
+func (conn *Connection) send(req Request, streamId uint32) *Future {
 	fut := conn.newFuture()
 	if fut.ready == nil {
 		return fut
 	}
-	conn.putFuture(fut, req)
+	conn.putFuture(fut, req, streamId)
 	return fut
 }
 
-func (conn *Connection) putFuture(fut *Future, req Request) {
+func (conn *Connection) putFuture(fut *Future, req Request, streamId uint32) {
 	shardn := fut.requestId & (conn.opts.Concurrency - 1)
 	shard := &conn.shard[shardn]
 	shard.bufmut.Lock()
@@ -811,7 +825,7 @@ func (conn *Connection) putFuture(fut *Future, req Request) {
 	}
 	blen := shard.buf.Len()
 	reqid := fut.requestId
-	if err := pack(&shard.buf, shard.enc, reqid, req, conn.Schema); err != nil {
+	if err := pack(&shard.buf, shard.enc, reqid, req, streamId, conn.Schema); err != nil {
 		shard.buf.Trunc(blen)
 		shard.bufmut.Unlock()
 		if f := conn.fetchFuture(reqid); f == fut {
@@ -993,7 +1007,7 @@ func (conn *Connection) nextRequestId() (requestId uint32) {
 // An error is returned if the request was formed incorrectly, or failed to
 // create the future.
 func (conn *Connection) Do(req Request) *Future {
-	return conn.send(req)
+	return conn.send(req, ignoreStreamId)
 }
 
 // ConfiguredTimeout returns a timeout from connection config.
@@ -1009,3 +1023,16 @@ func (conn *Connection) OverrideSchema(s *Schema) {
 		conn.Schema = s
 	}
 }
+
+// NewStream creates new Stream object for connection.
+//
+// Since v. 2.10.0, Tarantool supports streams and interactive transactions over them.
+// To use interactive transactions, memtx_use_mvcc_engine box option should be set to true.
+// Since 1.7.0
+func (conn *Connection) NewStream() *Stream {
+	atomic.AddUint32(&conn.lastStreamId, 1)
+	return &Stream{
+		Id:   conn.lastStreamId,
+		Conn: conn,
+	}
+}

connection_pool/connection_pool.go

@@ -534,6 +534,20 @@ func (connPool *ConnectionPool) Do(req tarantool.Request, userMode Mode) *tarant
 	return conn.Do(req)
 }
 
+// NewStream creates new Stream object for connection selected
+// by userMode from connPool.
+//
+// Since v. 2.10.0, Tarantool supports streams and interactive transactions over them.
+// To use interactive transactions, memtx_use_mvcc_engine box option should be set to true.
+// Since 1.7.0
+func (connPool *ConnectionPool) NewStream(userMode Mode) (*tarantool.Stream, error) {
+	conn, err := connPool.getNextConnection(userMode)
+	if err != nil {
+		return nil, err
+	}
+	return conn.NewStream(), nil
+}
+
 //
 // private
 //