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()`, `DoAsync()`, `DoTyped()`,
   `Begin()`, `Commit()`, `Rollback()` methods,
   `Begin()` - start transaction via iproto stream;
   `Commit()` - commit transaction;
   `Rollback()` - rollback transaction.

Closes #101

Author: AnaNek

CHANGELOG.md

@@ -36,6 +36,7 @@ CI and documentation.
 - queue-utube handling (#85)
 - Master discovery (#113)
 - SQL support (#62)
+- Streams and interactive transactions support (#101)
 
 ### Changed
 

connection.go

@@ -139,6 +139,8 @@ type Connection struct {
 	state   uint32
 	dec     *msgpack.Decoder
 	lenbuf  [PacketLengthBytes]byte
+
+	lastStreamId uint32
 }
 
 var _ = Connector(&Connection{}) // Check compatibility with connector interface.
@@ -472,11 +474,12 @@ func pack(h *smallWBuf, enc *msgpack.Encoder, reqid uint32, req Request, res Sch
 	hl := h.Len()
 	h.Write([]byte{
 		0xce, 0, 0, 0, 0, // Length.
-		0x82,                      // 2 element map.
+		0x83,                      // 3 element map.
 		KeyCode, byte(req.Code()), // Request code.
 		KeySync, 0xce,
 		byte(reqid >> 24), byte(reqid >> 16),
 		byte(reqid >> 8), byte(reqid),
+		KeyStreamId, byte(req.StreamId()),
 	})
 
 	if err = req.Body(res, enc); err != nil {
@@ -1027,3 +1030,15 @@ 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.
+func (conn *Connection) NewStream() *Stream {
+	conn.lastStreamId += 1
+	return &Stream{
+		Id:   conn.lastStreamId,
+		Conn: conn,
+	}
+}

connection_pool/connection_pool.go

@@ -512,6 +512,19 @@ func (connPool *ConnectionPool) DoAsync(req tarantool.Request, userMode Mode) *t
 	return conn.DoAsync(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.
+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
 //

connection_pool/connection_pool_test.go

@@ -1291,6 +1291,226 @@ func TestDo(t *testing.T) {
 	require.NotNilf(t, resp, "response is nil after Ping")
 }
 
+func TestStream_Commit(t *testing.T) {
+	var req tarantool.Request
+	var resp *tarantool.Response
+
+	// Tarantool supports streams and interactive transactions since version 2.10.0
+	isLess, err := test_helpers.IsTarantoolVersionLess(2, 10, 0)
+	if err != nil {
+		t.Fatal("Could not check the Tarantool version")
+	}
+	if isLess {
+		t.Skip()
+	}
+
+	roles := []bool{true, true, false, true, true}
+
+	err = test_helpers.SetClusterRO(servers, connOpts, roles)
+	require.Nilf(t, err, "fail to set roles for cluster")
+
+	connPool, err := connection_pool.Connect(servers, connOpts)
+	require.Nilf(t, err, "failed to connect")
+	require.NotNilf(t, connPool, "conn is nil after Connect")
+	defer connPool.Close()
+
+	stream, err := connPool.NewStream(connection_pool.PreferRW)
+	require.Nilf(t, err, "failed to create stream")
+	require.NotNilf(t, connPool, "stream is nil after NewStream")
+
+	// Begin transaction
+	resp, err = stream.Begin()
+	require.Nilf(t, err, "failed to Begin")
+	require.NotNilf(t, resp, "response is nil after Begin")
+	require.Equalf(t, uint32(0), resp.Code, "failed to Begin: wrong code returned")
+
+	// Insert in stream
+	req = tarantool.NewInsertRequest(spaceName).
+		Tuple([]interface{}{"commit_key", "commit_value"})
+	resp, err = stream.Do(req)
+	require.Nilf(t, err, "failed to Insert")
+	require.NotNilf(t, resp, "response is nil after Insert")
+	require.Equalf(t, uint32(0), resp.Code, "failed to Insert: wrong code returned")
+
+	// Connect to servers[2] to check if tuple
+	// was inserted on RW instance
+	conn, err := tarantool.Connect(servers[2], connOpts)
+	require.Nilf(t, err, "failed to connect %s", servers[2])
+	require.NotNilf(t, conn, "conn is nil after Connect")
+
+	// Select not related to the transaction
+	// while transaction is not commited
+	// result of select is empty
+	req = tarantool.NewSelectRequest(spaceNo).
+		Index(indexNo).
+		Offset(0).
+		Limit(1).
+		Iterator(tarantool.IterEq).
+		Key([]interface{}{"commit_key"})
+	resp, err = conn.Do(req)
+	require.Nilf(t, err, "failed to Select")
+	require.NotNilf(t, resp, "response is nil after Select")
+	require.Equalf(t, 0, len(resp.Data), "response Data len != 0")
+
+	// Select in stream
+	req = tarantool.NewSelectRequest(spaceNo).
+		Index(indexNo).
+		Offset(0).
+		Limit(1).
+		Iterator(tarantool.IterEq).
+		Key([]interface{}{"commit_key"})
+	resp, err = stream.Do(req)
+	require.Nilf(t, err, "failed to Select")
+	require.NotNilf(t, resp, "response is nil after Select")
+	require.Equalf(t, 1, len(resp.Data), "response Body len != 1 after Select")
+
+	tpl, ok := resp.Data[0].([]interface{})
+	require.Truef(t, ok, "unexpected body of Select")
+	require.Equalf(t, 2, len(tpl), "unexpected body of Select")
+
+	key, ok := tpl[0].(string)
+	require.Truef(t, ok, "unexpected body of Select (0)")
+	require.Equalf(t, "commit_key", key, "unexpected body of Select (0)")
+
+	value, ok := tpl[1].(string)
+	require.Truef(t, ok, "unexpected body of Select (1)")
+	require.Equalf(t, "commit_value", value, "unexpected body of Select (1)")
+
+	// Commit transaction
+	resp, err = stream.Commit()
+	require.Nilf(t, err, "failed to Commit")
+	require.NotNilf(t, resp, "response is nil after Commit")
+	require.Equalf(t, uint32(0), resp.Code, "failed to Commit: wrong code returned")
+
+	// Select outside of transaction
+	req = tarantool.NewSelectRequest(spaceNo).
+		Index(indexNo).
+		Offset(0).
+		Limit(1).
+		Iterator(tarantool.IterEq).
+		Key([]interface{}{"commit_key"})
+	resp, err = conn.Do(req)
+	require.Nilf(t, err, "failed to Select")
+	require.NotNilf(t, resp, "response is nil after Select")
+	require.Equalf(t, len(resp.Data), 1, "response Body len != 1 after Select")
+
+	tpl, ok = resp.Data[0].([]interface{})
+	require.Truef(t, ok, "unexpected body of Select")
+	require.Equalf(t, 2, len(tpl), "unexpected body of Select")
+
+	key, ok = tpl[0].(string)
+	require.Truef(t, ok, "unexpected body of Select (0)")
+	require.Equalf(t, "commit_key", key, "unexpected body of Select (0)")
+
+	value, ok = tpl[1].(string)
+	require.Truef(t, ok, "unexpected body of Select (1)")
+	require.Equalf(t, "commit_value", value, "unexpected body of Select (1)")
+}
+
+func TestStream_Rollback(t *testing.T) {
+	var req tarantool.Request
+	var resp *tarantool.Response
+
+	// Tarantool supports streams and interactive transactions since version 2.10.0
+	isLess, err := test_helpers.IsTarantoolVersionLess(2, 10, 0)
+	if err != nil {
+		t.Fatal("Could not check the Tarantool version")
+	}
+	if isLess {
+		t.Skip()
+	}
+
+	roles := []bool{true, true, false, true, true}
+
+	err = test_helpers.SetClusterRO(servers, connOpts, roles)
+	require.Nilf(t, err, "fail to set roles for cluster")
+
+	connPool, err := connection_pool.Connect(servers, connOpts)
+	require.Nilf(t, err, "failed to connect")
+	require.NotNilf(t, connPool, "conn is nil after Connect")
+	defer connPool.Close()
+
+	stream, err := connPool.NewStream(connection_pool.PreferRW)
+	require.Nilf(t, err, "failed to create stream")
+	require.NotNilf(t, connPool, "stream is nil after NewStream")
+
+	// Begin transaction
+	resp, err = stream.Begin()
+	require.Nilf(t, err, "failed to Begin")
+	require.NotNilf(t, resp, "response is nil after Begin")
+	require.Equalf(t, uint32(0), resp.Code, "failed to Begin: wrong code returned")
+
+	// Insert in stream
+	req = tarantool.NewInsertRequest(spaceName).
+		Tuple([]interface{}{"rollback_key", "rollback_value"})
+	resp, err = stream.Do(req)
+	require.Nilf(t, err, "failed to Insert")
+	require.NotNilf(t, resp, "response is nil after Insert")
+	require.Equalf(t, uint32(0), resp.Code, "failed to Insert: wrong code returned")
+
+	// Connect to servers[2] to check if tuple
+	// was inserted on RW instance
+	conn, err := tarantool.Connect(servers[2], connOpts)
+	require.Nilf(t, err, "failed to connect %s", servers[2])
+	require.NotNilf(t, conn, "conn is nil after Connect")
+
+	// Select not related to the transaction
+	// while transaction is not commited
+	// result of select is empty
+	req = tarantool.NewSelectRequest(spaceNo).
+		Index(indexNo).
+		Offset(0).
+		Limit(1).
+		Iterator(tarantool.IterEq).
+		Key([]interface{}{"rollback_key"})
+	resp, err = conn.Do(req)
+	require.Nilf(t, err, "failed to Select")
+	require.NotNilf(t, resp, "response is nil after Select")
+	require.Equalf(t, 0, len(resp.Data), "response Data len != 0")
+
+	// Select in stream
+	req = tarantool.NewSelectRequest(spaceNo).
+		Index(indexNo).
+		Offset(0).
+		Limit(1).
+		Iterator(tarantool.IterEq).
+		Key([]interface{}{"rollback_key"})
+	resp, err = stream.Do(req)
+	require.Nilf(t, err, "failed to Select")
+	require.NotNilf(t, resp, "response is nil after Select")
+	require.Equalf(t, 1, len(resp.Data), "response Body len != 1 after Select")
+
+	tpl, ok := resp.Data[0].([]interface{})
+	require.Truef(t, ok, "unexpected body of Select")
+	require.Equalf(t, 2, len(tpl), "unexpected body of Select")
+
+	key, ok := tpl[0].(string)
+	require.Truef(t, ok, "unexpected body of Select (0)")
+	require.Equalf(t, "rollback_key", key, "unexpected body of Select (0)")
+
+	value, ok := tpl[1].(string)
+	require.Truef(t, ok, "unexpected body of Select (1)")
+	require.Equalf(t, "rollback_value", value, "unexpected body of Select (1)")
+
+	// Rollback transaction
+	resp, err = stream.Rollback()
+	require.Nilf(t, err, "failed to Rollback")
+	require.NotNilf(t, resp, "response is nil after Rollback")
+	require.Equalf(t, uint32(0), resp.Code, "failed to Rollback: wrong code returned")
+
+	// Select outside of transaction
+	req = tarantool.NewSelectRequest(spaceNo).
+		Index(indexNo).
+		Offset(0).
+		Limit(1).
+		Iterator(tarantool.IterEq).
+		Key([]interface{}{"rollback_key"})
+	resp, err = conn.Do(req)
+	require.Nilf(t, err, "failed to Select")
+	require.NotNilf(t, resp, "response is nil after Select")
+	require.Equalf(t, 0, len(resp.Data), "response Body len != 1 after Select")
+}
+
 // runTestMain is a body of TestMain function
 // (see https://pkg.go.dev/testing#hdr-Main).
 // Using defer + os.Exit is not works so TestMain body
@@ -1308,12 +1528,13 @@ func runTestMain(m *testing.M) int {
 	var err error
 
 	instances, err = test_helpers.StartTarantoolInstances(servers, workDirs, test_helpers.StartOpts{
-		InitScript:   initScript,
-		User:         connOpts.User,
-		Pass:         connOpts.Pass,
-		WaitStart:    waitStart,
-		ConnectRetry: connectRetry,
-		RetryTimeout: retryTimeout,
+		InitScript:         initScript,
+		User:               connOpts.User,
+		Pass:               connOpts.Pass,
+		WaitStart:          waitStart,
+		ConnectRetry:       connectRetry,
+		RetryTimeout:       retryTimeout,
+		MemtxUseMvccEngine: true,
 	})
 
 	if err != nil {

const.go

@@ -12,11 +12,15 @@ const (
 	UpsertRequestCode    = 9
 	Call17RequestCode    = 10
 	ExecuteRequestCode   = 11
+	BeginRequestCode     = 14
+	CommitRequestCode    = 15
+	RollbackRequestCode  = 16
 	PingRequestCode      = 64
 	SubscribeRequestCode = 66
 
 	KeyCode         = 0x00
 	KeySync         = 0x01
+	KeyStreamId     = 0x0a
 	KeySpaceNo      = 0x10
 	KeyIndexNo      = 0x11
 	KeyLimit        = 0x12