TNTP-3334: IPROTO_IS_SYNC support for begin/commit

CHANGELOG.md

@@ -14,6 +14,8 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 - Implemented box.session.su request and sugar interface only for current session granting (#426).
 - Defined `ErrConcurrentSchemaUpdate` constant for "concurrent schema update" error.
   Now you can check this error with `errors.Is(err, tarantool.ErrConcurrentSchemaUpdate)`.
+- Implemented support for `IPROTO_IS_SYNC` flag in stream transactions,
+  added `IsSync(bool)` method for `BeginRequest`/`CommitRequest` (#447).
 
 ### Changed
 

example_test.go

@@ -998,6 +998,90 @@ func ExampleBeginRequest_TxnIsolation() {
 	fmt.Printf("Select after Rollback: response is %#v\n", data)
 }
 
+func ExampleBeginRequest_IsSync() {
+	conn := exampleConnect(dialer, opts)
+	defer conn.Close()
+
+	// Tarantool supports IS_SYNC flag for BeginRequest since version 3.1.0.
+	isLess, err := test_helpers.IsTarantoolVersionLess(3, 1, 0)
+	if err != nil || isLess {
+		return
+	}
+
+	stream, err := conn.NewStream()
+	if err != nil {
+		fmt.Printf("error getting the stream: %s\n", err)
+		return
+	}
+
+	// Begin transaction with synchronous mode
+	req := tarantool.NewBeginRequest().IsSync(true)
+	resp, err := stream.Do(req).GetResponse()
+	switch {
+	case err != nil:
+		fmt.Printf("error getting the response: %s\n", err)
+	case resp.Header().Error != tarantool.ErrorNo:
+		fmt.Printf("response error code: %s\n", resp.Header().Error)
+	default:
+		fmt.Println("Success.")
+	}
+}
+
+func ExampleCommitRequest_IsSync() {
+	conn := exampleConnect(dialer, opts)
+	defer conn.Close()
+
+	// Tarantool supports IS_SYNC flag for CommitRequest since version 3.1.0.
+	isLess, err := test_helpers.IsTarantoolVersionLess(3, 1, 0)
+	if err != nil || isLess {
+		return
+	}
+
+	var req tarantool.Request
+
+	stream, err := conn.NewStream()
+	if err != nil {
+		fmt.Printf("error getting the stream: %s\n", err)
+		return
+	}
+
+	// Begin transaction.
+	req = tarantool.NewBeginRequest()
+	resp, err := stream.Do(req).GetResponse()
+	switch {
+	case err != nil:
+		fmt.Printf("error getting the response: %s\n", err)
+		return
+	case resp.Header().Error != tarantool.ErrorNo:
+		fmt.Printf("response error code: %s\n", resp.Header().Error)
+		return
+	}
+
+	// Insert in stream.
+	req = tarantool.NewReplaceRequest("test").Tuple([]interface{}{1, "test"})
+	resp, err = stream.Do(req).GetResponse()
+	switch {
+	case err != nil:
+		fmt.Printf("error getting the response: %s\n", err)
+		return
+	case resp.Header().Error != tarantool.ErrorNo:
+		fmt.Printf("response error code: %s\n", resp.Header().Error)
+		return
+	}
+
+	// Commit transaction in sync mode.
+	req = tarantool.NewCommitRequest().IsSync(true)
+	resp, err = stream.Do(req).GetResponse()
+	switch {
+	case err != nil:
+		fmt.Printf("error getting the response: %s\n", err)
+	case resp.Header().Error != tarantool.ErrorNo:
+		fmt.Printf("response error code: %s\n", resp.Header().Error)
+	default:
+		fmt.Println("Success.")
+	}
+}
+
 func ExampleErrorNo() {
 	conn := exampleConnect(dialer, opts)
 	defer conn.Close()

prepared.go

@@ -22,26 +22,6 @@ type Prepared struct {
 	Conn        *Connection
 }
 
-func fillPrepare(enc *msgpack.Encoder, expr string) error {
-	enc.EncodeMapLen(1)
-	enc.EncodeUint(uint64(iproto.IPROTO_SQL_TEXT))
-	return enc.EncodeString(expr)
-}
-
-func fillUnprepare(enc *msgpack.Encoder, stmt Prepared) error {
-	enc.EncodeMapLen(1)
-	enc.EncodeUint(uint64(iproto.IPROTO_STMT_ID))
-	return enc.EncodeUint(uint64(stmt.StatementID))
-}
-
-func fillExecutePrepared(enc *msgpack.Encoder, stmt Prepared, args interface{}) error {
-	enc.EncodeMapLen(2)
-	enc.EncodeUint(uint64(iproto.IPROTO_STMT_ID))
-	enc.EncodeUint(uint64(stmt.StatementID))
-	enc.EncodeUint(uint64(iproto.IPROTO_SQL_BIND))
-	return encodeSQLBind(enc, args)
-}
-
 // NewPreparedFromResponse constructs a Prepared object.
 func NewPreparedFromResponse(conn *Connection, resp Response) (*Prepared, error) {
 	if resp == nil {
@@ -81,8 +61,16 @@ func NewPrepareRequest(expr string) *PrepareRequest {
 }
 
 // Body fills an msgpack.Encoder with the execute request body.
-func (req *PrepareRequest) Body(res SchemaResolver, enc *msgpack.Encoder) error {
-	return fillPrepare(enc, req.expr)
+func (req *PrepareRequest) Body(_ SchemaResolver, enc *msgpack.Encoder) error {
+	err := enc.EncodeMapLen(1)
+	if err != nil {
+		return err
+	}
+	err = enc.EncodeUint(uint64(iproto.IPROTO_SQL_TEXT))
+	if err != nil {
+		return err
+	}
+	return enc.EncodeString(req.expr)
 }
 
 // Context sets a passed context to the request.
@@ -126,8 +114,16 @@ func (req *UnprepareRequest) Conn() *Connection {
 }
 
 // Body fills an msgpack.Encoder with the execute request body.
-func (req *UnprepareRequest) Body(res SchemaResolver, enc *msgpack.Encoder) error {
-	return fillUnprepare(enc, *req.stmt)
+func (req *UnprepareRequest) Body(_ SchemaResolver, enc *msgpack.Encoder) error {
+	err := enc.EncodeMapLen(1)
+	if err != nil {
+		return err
+	}
+	err = enc.EncodeUint(uint64(iproto.IPROTO_STMT_ID))
+	if err != nil {
+		return err
+	}
+	return enc.EncodeUint(uint64(req.stmt.StatementID))
 }
 
 // Context sets a passed context to the request.
@@ -171,8 +167,24 @@ func (req *ExecutePreparedRequest) Args(args interface{}) *ExecutePreparedReques
 }
 
 // Body fills an msgpack.Encoder with the execute request body.
-func (req *ExecutePreparedRequest) Body(res SchemaResolver, enc *msgpack.Encoder) error {
-	return fillExecutePrepared(enc, *req.stmt, req.args)
+func (req *ExecutePreparedRequest) Body(_ SchemaResolver, enc *msgpack.Encoder) error {
+	err := enc.EncodeMapLen(2)
+	if err != nil {
+		return err
+	}
+	err = enc.EncodeUint(uint64(iproto.IPROTO_STMT_ID))
+	if err != nil {
+		return err
+	}
+	err = enc.EncodeUint(uint64(req.stmt.StatementID))
+	if err != nil {
+		return err
+	}
+	err = enc.EncodeUint(uint64(iproto.IPROTO_SQL_BIND))
+	if err != nil {
+		return err
+	}
+	return encodeSQLBind(enc, req.args)
 }
 
 // Context sets a passed context to the request.