stream: added is_sync flag for begin/commit

Author: bigbes

internal/request/begin_test.go

@@ -0,0 +1,68 @@
+package request
+
+import (
+	"bytes"
+	"fmt"
+	"github.com/stretchr/testify/require"
+	"github.com/tarantool/go-tarantool/v2"
+	"github.com/vmihailenco/msgpack/v5"
+	"testing"
+	"time"
+
+	"github.com/kr/pretty"
+)
+
+/*
+example file:
+
+BEGIN {
+	"Timeout": 0.123,
+	"TxnIsolation": 1,
+ 	"IsSync": true,
+}
+EXPECTED "8356CB3F847AE147AE147B590161C3"
+
+COMMIT {
+	"
+}
+*/
+
+func CompareEncodeDecode(t *testing.T, expectedMsgpack []byte, gotMsgpack []byte) bool {
+	t.Helper()
+
+	tarantool.NewBeginRequest()
+
+	return bytes.Equal(expectedMsgpack, gotMsgpack)
+}
+
+type dummySchemaResolver struct{}
+
+func (d dummySchemaResolver) ResolveSpace(interface{}) (uint32, error)         { panic("unexpected") }
+func (d dummySchemaResolver) ResolveIndex(interface{}, uint32) (uint32, error) { panic("unexpected") }
+func (d dummySchemaResolver) NamesUseSupported() bool                          { panic("unexpected") }
+
+func TestBegin(t *testing.T) {
+	var b bytes.Buffer
+	encoder := msgpack.NewEncoder(&b)
+
+	req := tarantool.NewBeginRequest().TxnIsolation(1).IsSync(true).Timeout(10 * time.Millisecond)
+	err := req.Body(dummySchemaResolver{}, encoder)
+	require.NoError(t, err)
+
+	dumpBinary(t, b.Bytes())
+
+	var abstractRequest map[int]interface{}
+
+	err = msgpack.NewDecoder(&b).Decode(&abstractRequest)
+	require.NoError(t, err)
+
+	pretty.Println(abstractRequest)
+}
+
+func dumpBinary(t *testing.T, b []byte) {
+	t.Helper()
+
+	for _, v := range b {
+		fmt.Printf("%02X", v)
+	}
+}

internal/utils/optional.go

@@ -0,0 +1,22 @@
+package utils
+
+type Optional[T any] struct {
+	isSet bool
+	value T
+}
+
+func (o Optional[T]) Value() (T, bool) {
+	return o.value, o.isSet
+}
+
+func (o Optional[T]) HasValue() bool {
+	return o.isSet
+}
+
+func Some[T any](val T) Optional[T] {
+	return Optional[T]{value: val, isSet: true}
+}
+
+func None[T any]() Optional[T] {
+	return Optional[T]{} // IsSet is false by default
+}

stream.go

@@ -3,6 +3,7 @@ package tarantool
 import (
 	"context"
 	"errors"
+	"github.com/tarantool/go-tarantool/v2/internal/utils"
 	"time"
 
 	"github.com/tarantool/go-iproto"
@@ -42,6 +43,7 @@ type BeginRequest struct {
 	baseRequest
 	txnIsolation TxnIsolationLevel
 	timeout      time.Duration
+	isSync       utils.Optional[bool]
 }
 
 // NewBeginRequest returns a new BeginRequest.
@@ -59,12 +61,18 @@ func (req *BeginRequest) TxnIsolation(txnIsolation TxnIsolationLevel) *BeginRequ
 	return req
 }
 
-// WithTimeout allows to set up a timeout for call BeginRequest.
+// Timeout allows to set up a timeout for call BeginRequest.
 func (req *BeginRequest) Timeout(timeout time.Duration) *BeginRequest {
 	req.timeout = timeout
 	return req
 }
 
+// IsSync allows to set up a IsSync flag for call BeginRequest.
+func (req *BeginRequest) IsSync(isSync bool) *BeginRequest {
+	req.isSync = utils.Some(isSync)
+	return req
+}
+
 // Body fills an msgpack.Encoder with the begin request body.
 func (req *BeginRequest) Body(_ SchemaResolver, enc *msgpack.Encoder) error {
 	var (
@@ -81,6 +89,10 @@ func (req *BeginRequest) Body(_ SchemaResolver, enc *msgpack.Encoder) error {
 		mapLen++
 	}
 
+	if req.isSync.HasValue() {
+		mapLen++
+	}
+
 	err := enc.EncodeMapLen(mapLen)
 	if err != nil {
 		return err
@@ -110,6 +122,18 @@ func (req *BeginRequest) Body(_ SchemaResolver, enc *msgpack.Encoder) error {
 		}
 	}
 
+	if val, ok := req.isSync.Value(); ok {
+		err = enc.EncodeUint(uint64(iproto.IPROTO_IS_SYNC))
+		if err != nil {
+			return err
+		}
+
+		err = enc.EncodeBool(val)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -129,6 +153,8 @@ func (req *BeginRequest) Context(ctx context.Context) *BeginRequest {
 // Commit request can not be processed out of stream.
 type CommitRequest struct {
 	baseRequest
+
+	isSync utils.Optional[bool]
 }
 
 // NewCommitRequest returns a new CommitRequest.
@@ -138,9 +164,37 @@ func NewCommitRequest() *CommitRequest {
 	return req
 }
 
+// IsSync allows to set up a IsSync flag for call BeginRequest.
+func (req *CommitRequest) IsSync(isSync bool) *CommitRequest {
+	req.isSync = utils.Some(isSync)
+	return req
+}
+
 // Body fills an msgpack.Encoder with the commit request body.
 func (req *CommitRequest) Body(res SchemaResolver, enc *msgpack.Encoder) error {
-	return enc.EncodeMapLen(0)
+	var (
+		mapLen = 0
+	)
+
+	if req.isSync.HasValue() {
+		mapLen++
+	}
+
+	if err := enc.EncodeMapLen(mapLen); err != nil {
+		return err
+	}
+
+	if val, ok := req.isSync.Value(); ok {
+		if err := enc.EncodeUint(uint64(iproto.IPROTO_IS_SYNC)); err != nil {
+			return err
+		}
+
+		if err := enc.EncodeBool(val); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 // Context sets a passed context to the request.

tarantool_test.go

@@ -4134,6 +4134,40 @@ func TestFdDialer(t *testing.T) {
 	require.Equal(t, int8(0), resp[0])
 }
 
+func TestIsSync(t *testing.T) {
+	test_helpers.SkipIfIsSyncUnsupported(t)
+
+	conn := test_helpers.ConnectWithValidation(t, dialer, opts)
+	defer conn.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	stream, err := conn.NewStream()
+	require.NoError(t, err)
+
+	requests := []Request{
+		NewBeginRequest().IsSync(true),
+		NewInsertRequest("test").Tuple([]interface{}{1, "test"}),
+		NewCommitRequest().IsSync(true),
+		//NewBeginRequest(),
+		//NewInsertRequest("test").Tuple([]interface{}{1, "test"}),
+		//NewCommitRequest(),
+	}
+
+	for _, req := range requests {
+		status := stream.Do(req)
+
+		select {
+		case <-status.WaitChan():
+			_, err := status.Get()
+			require.NoError(t, err)
+		case <-ctx.Done():
+			require.Fail(t, "request failed", ctx.Err().Error())
+		}
+	}
+}
+
 // 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

test_helpers/utils.go

@@ -217,6 +217,14 @@ func SkipIfCrudSpliceBroken(t *testing.T) {
 	SkipIfFeatureUnsupported(t, "crud update splice", 2, 0, 0)
 }
 
+// SkipIfIsSyncUnsupported skips test run if Tarantool without
+// IS_SYNC support is used.
+func SkipIfIsSyncUnsupported(t *testing.T) {
+	t.Helper()
+
+	SkipIfFeatureUnsupported(t, "is sync", 3, 1, 0)
+}
+
 // IsTcsSupported checks if Tarantool supports centralized storage.
 // Tarantool supports centralized storage with Enterprise since 3.3.0 version.
 func IsTcsSupported() (bool, error) {