api: create different responses for different requests

Different implementations of the `Response` interface created.
Special types of responses are used with special requests.

Thus `Response` interface was simplified: some special methods
were moved to the corresponding implementations.
This means that if a user wants to access this methods, the
response should be casted to its actual type.

`SelectResponse`, `ExecuteResponse`, `PrepareResponse`,
`PushResponse` are part of a public API. `Pos()`, `MetaData()`,
`SQLInfo()` methods created for them to get specific info.

`Future` constructors now accept `Request` as their argument.
`Future` methods `AppendPush` and `SetResponse` accept
response `Header` and data as their arguments.

`IsPush()` method is used to return the information if the current
response is a `PushResponse`. `PushCode` constant is removed. To get
information, if the current response is a push response, `IsPush()`
method could be used instead.

After this patch, operations `Ping`, `Select`, `Insert`, `Replace`,
`Delete`, `Update`, `Upsert`, `Call`, `Call16`, `Call17`, `Eval`, `Execute`
of a `Connector` return response data instead of an actual responses.
After this patch, operations `Ping`, `Select`, `Insert`, `Replace`,
`Delete`, `Update`, `Upsert`, `Call`, `Call16`, `Call17`, `Eval`, `Execute`
of a `Pooler` return response data instead of an actual responses.

Part of #237

Author: DerekBum

CHANGELOG.md

@@ -30,6 +30,9 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 - Support connection via an existing socket fd (#321)
 - `Header` struct for the response header (#237). It can be accessed via
   `Header()` method of the `Response` interface.
+- `Response` method added to the `Request` interface (#237)
+- New `LogAppendPushFailed` connection log constant (#237).
+  It is logged when connection fails to append a push response.
 
 ### Changed
 
@@ -70,12 +73,21 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 - Rename `pool.GetPoolInfo` to `pool.GetInfo`. Change return type to
   `map[string]ConnectionInfo` (#321)
 - `Response` is now an interface (#237)
-- All responses are now implementations of the `Response` interface (#237)
+- All responses are now implementations of the `Response` interface (#237).
+  `SelectResponse`, `ExecuteResponse`, `PrepareResponse`, `PushResponse` are part
+  of a public API. `Pos()`, `MetaData()`, `SQLInfo()` methods created for them
+  to get specific info.
+  Special types of responses are used with special requests.
 - `IsPush()` method is added to the response iterator (#237). It returns
   the information if the current response is a `PushResponse`.
   `PushCode` constant is removed.
 - Method `Get` for `Future` now returns response data (#237). To get the actual
-  response new `GetResponse` method has been added.
+  response new `GetResponse` method has been added. Methods `AppendPush` and
+  `SetResponse` accept response `Header` and data as their arguments.
+- `Future` constructors now accept `Request` as their argument (#237)
+- Operations `Ping`, `Select`, `Insert`, `Replace`, `Delete`, `Update`, `Upsert`,
+  `Call`, `Call16`, `Call17`, `Eval`, `Execute` of a `Connector` and `Pooler`
+  return response data instead of an actual responses (#237)
 
 ### Deprecated
 

README.md

@@ -202,6 +202,9 @@ The subpackage has been deleted. You could use `pool` instead.
   unique string ID, which allows them to be distinguished.
 * `pool.GetPoolInfo` has been renamed to `pool.GetInfo`. Return type has been changed
   to `map[string]ConnectionInfo`.
+* Operations `Ping`, `Select`, `Insert`, `Replace`, `Delete`, `Update`, `Upsert`,
+  `Call`, `Call16`, `Call17`, `Eval`, `Execute` of a `Pooler` return
+  response data instead of an actual responses.
 
 #### crud package
 
@@ -257,6 +260,7 @@ longer accept `ops` argument (operations) as an `interface{}`. `*Operations`
 needs to be passed instead.
 * `UpdateRequest` and `UpsertRequest` structs no longer accept `interface{}`
 for an `ops` field. `*Operations` needs to be used instead.
+* `Response` method added to the `Request` interface.
 
 #### Response changes
 
@@ -265,11 +269,25 @@ for an `ops` field. `*Operations` needs to be used instead.
   `Header()` method.
 * `ResponseIterator` interface now has `IsPush()` method.
   It returns true if the current response is a push response.
+* For each request type, a different response type is created. They all
+  implement a `Response` interface. `SelectResponse`, `PrepareResponse`,
+  `ExecuteResponse`, `PushResponse` are a part of a public API.
+  `Pos()`, `MetaData()`, `SQLInfo()` methods created for them to get specific info.
+  Special types of responses are used with special requests.
 
 #### Future changes
 
 * Method `Get` now returns response data instead of the actual response.
 * New method `GetResponse` added to get an actual response.
+* `Future` constructors now accept `Request` as their argument.
+* Methods `AppendPush` and `SetResponse` accepts response `Header` and data
+  as their arguments.
+
+#### Connector changes
+
+Operations `Ping`, `Select`, `Insert`, `Replace`, `Delete`, `Update`, `Upsert`, 
+`Call`, `Call16`, `Call17`, `Eval`, `Execute` of a `Connector` return 
+response data instead of an actual responses.
 
 #### Connect function
 

box_error_test.go

@@ -304,9 +304,7 @@ func TestErrorTypeEval(t *testing.T) {
 
 	for name, testcase := range tupleCases {
 		t.Run(name, func(t *testing.T) {
-			resp, err := conn.Eval("return ...", []interface{}{&testcase.tuple.val})
-			require.Nil(t, err)
-			data, err := resp.Decode()
+			data, err := conn.Eval("return ...", []interface{}{&testcase.tuple.val})
 			require.Nil(t, err)
 			require.NotNil(t, data)
 			require.Equal(t, len(data), 1)
@@ -438,14 +436,11 @@ func TestErrorTypeSelect(t *testing.T) {
 			_, err := conn.Eval(insertEval, []interface{}{})
 			require.Nilf(t, err, "Tuple has been successfully inserted")
 
-			var resp Response
 			var offset uint32 = 0
 			var limit uint32 = 1
-			resp, err = conn.Select(space, index, offset, limit, IterEq,
+			data, err := conn.Select(space, index, offset, limit, IterEq,
 				[]interface{}{testcase.tuple.pk})
 			require.Nil(t, err)
-			data, err := resp.Decode()
-			require.Nil(t, err)
 			require.NotNil(t, data)
 			require.Equalf(t, len(data), 1, "Exactly one tuple had been found")
 			tpl, ok := data[0].([]interface{})

connection.go

@@ -61,6 +61,8 @@ const (
 	LogUnexpectedResultId
 	// LogWatchEventReadFailed is logged when failed to read a watch event.
 	LogWatchEventReadFailed
+	// LogAppendPushFailed is logged when failed to append a push response.
+	LogAppendPushFailed
 )
 
 // ConnEvent is sent throw Notify channel specified in Opts.
@@ -103,6 +105,9 @@ func (d defaultLogger) Report(event ConnLogKind, conn *Connection, v ...interfac
 	case LogWatchEventReadFailed:
 		err := v[0].(error)
 		log.Printf("tarantool: unable to parse watch event: %s", err)
+	case LogAppendPushFailed:
+		err := v[0].(error)
+		log.Printf("tarantool: unable to append a push response: %s", err)
 	default:
 		args := append([]interface{}{"tarantool: unexpected event ", event, conn}, v...)
 		log.Print(args...)
@@ -818,10 +823,9 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 			return
 		}
 
-		resp := &ConnResponse{header: header, buf: buf}
 		var fut *Future = nil
 		if iproto.Type(header.Code) == iproto.IPROTO_EVENT {
-			if event, err := readWatchEvent(&resp.buf); err == nil {
+			if event, err := readWatchEvent(&buf); err == nil {
 				events <- event
 			} else {
 				err = ClientError{
@@ -833,11 +837,19 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 			continue
 		} else if header.Code == uint32(iproto.IPROTO_CHUNK) {
 			if fut = conn.peekFuture(header.RequestId); fut != nil {
-				fut.AppendPush(resp)
+				if err := fut.AppendPush(header, &buf); err != nil {
+					err = ClientError{
+						ErrProtocolError,
+						fmt.Sprintf("failed to append push response: %s", err),
+					}
+					conn.opts.Logger.Report(LogAppendPushFailed, conn, err)
+				}
 			}
 		} else {
 			if fut = conn.fetchFuture(header.RequestId); fut != nil {
-				fut.SetResponse(resp)
+				if err := fut.SetResponse(header, &buf); err != nil {
+					fut.SetError(fmt.Errorf("failed to set response: %w", err))
+				}
 				conn.markDone(fut)
 			}
 		}
@@ -873,8 +885,10 @@ func (conn *Connection) eventer(events <-chan connWatchEvent) {
 	}
 }
 
-func (conn *Connection) newFuture(ctx context.Context) (fut *Future) {
+func (conn *Connection) newFuture(req Request) (fut *Future) {
+	ctx := req.Ctx()
 	fut = NewFuture()
+	fut.SetRequest(req)
 	if conn.rlimit != nil && conn.opts.RLimitAction == RLimitDrop {
 		select {
 		case conn.rlimit <- struct{}{}:
@@ -984,7 +998,7 @@ func (conn *Connection) decrementRequestCnt() {
 func (conn *Connection) send(req Request, streamId uint64) *Future {
 	conn.incrementRequestCnt()
 
-	fut := conn.newFuture(req.Ctx())
+	fut := conn.newFuture(req)
 	if fut.ready == nil {
 		conn.decrementRequestCnt()
 		return fut
@@ -1053,8 +1067,11 @@ func (conn *Connection) putFuture(fut *Future, req Request, streamId uint64) {
 
 	if req.Async() {
 		if fut = conn.fetchFuture(reqid); fut != nil {
-			resp := &ConnResponse{}
-			fut.SetResponse(resp)
+			header := Header{
+				RequestId: reqid,
+				Code:      OkCode,
+			}
+			fut.SetResponse(header, nil)
 			conn.markDone(fut)
 		}
 	}

connector.go

@@ -13,41 +13,41 @@ type Connector interface {
 
 	// Deprecated: the method will be removed in the next major version,
 	// use a PingRequest object + Do() instead.
-	Ping() (Response, error)
+	Ping() ([]interface{}, error)
 	// Deprecated: the method will be removed in the next major version,
 	// use a SelectRequest object + Do() instead.
 	Select(space, index interface{}, offset, limit uint32, iterator Iter,
-		key interface{}) (Response, error)
+		key interface{}) ([]interface{}, error)
 	// Deprecated: the method will be removed in the next major version,
 	// use an InsertRequest object + Do() instead.
-	Insert(space interface{}, tuple interface{}) (Response, error)
+	Insert(space interface{}, tuple interface{}) ([]interface{}, error)
 	// Deprecated: the method will be removed in the next major version,
 	// use a ReplicaRequest object + Do() instead.
-	Replace(space interface{}, tuple interface{}) (Response, error)
+	Replace(space interface{}, tuple interface{}) ([]interface{}, error)
 	// Deprecated: the method will be removed in the next major version,
 	// use a DeleteRequest object + Do() instead.
-	Delete(space, index interface{}, key interface{}) (Response, error)
+	Delete(space, index interface{}, key interface{}) ([]interface{}, error)
 	// Deprecated: the method will be removed in the next major version,
 	// use a UpdateRequest object + Do() instead.
-	Update(space, index interface{}, key interface{}, ops *Operations) (Response, error)
+	Update(space, index interface{}, key interface{}, ops *Operations) ([]interface{}, error)
 	// Deprecated: the method will be removed in the next major version,
 	// use a UpsertRequest object + Do() instead.
-	Upsert(space interface{}, tuple interface{}, ops *Operations) (Response, error)
+	Upsert(space interface{}, tuple interface{}, ops *Operations) ([]interface{}, error)
 	// Deprecated: the method will be removed in the next major version,
 	// use a CallRequest object + Do() instead.
-	Call(functionName string, args interface{}) (Response, error)
+	Call(functionName string, args interface{}) ([]interface{}, error)
 	// Deprecated: the method will be removed in the next major version,
 	// use a Call16Request object + Do() instead.
-	Call16(functionName string, args interface{}) (Response, error)
+	Call16(functionName string, args interface{}) ([]interface{}, error)
 	// Deprecated: the method will be removed in the next major version,
 	// use a Call17Request object + Do() instead.
-	Call17(functionName string, args interface{}) (Response, error)
+	Call17(functionName string, args interface{}) ([]interface{}, error)
 	// Deprecated: the method will be removed in the next major version,
 	// use an EvalRequest object + Do() instead.
-	Eval(expr string, args interface{}) (Response, error)
+	Eval(expr string, args interface{}) ([]interface{}, error)
 	// Deprecated: the method will be removed in the next major version,
 	// use an ExecuteRequest object + Do() instead.
-	Execute(expr string, args interface{}) (Response, error)
+	Execute(expr string, args interface{}) ([]interface{}, error)
 
 	// Deprecated: the method will be removed in the next major version,
 	// use a SelectRequest object + Do() instead.

crud/common.go

@@ -55,6 +55,7 @@ package crud
 
 import (
 	"context"
+	"io"
 
 	"github.com/tarantool/go-iproto"
 
@@ -84,6 +85,12 @@ func (req baseRequest) Async() bool {
 	return req.impl.Async()
 }
 
+// Response creates a response for the baseRequest.
+func (req baseRequest) Response(header tarantool.Header,
+	body io.Reader) (tarantool.Response, error) {
+	return req.impl.Response(header, body)
+}
+
 type spaceRequest struct {
 	baseRequest
 	space string

dial.go

@@ -404,16 +404,11 @@ func identify(w writeFlusher, r io.Reader) (ProtocolInfo, error) {
 	}
 	data, err := resp.Decode()
 	if err != nil {
-		switch err := err.(type) {
-		case Error:
-			if err.Code == iproto.ER_UNKNOWN_REQUEST_TYPE {
-				// IPROTO_ID requests are not supported by server.
-				return info, nil
-			}
-			return info, err
-		default:
-			return info, fmt.Errorf("decode response body error: %w", err)
+		if iproto.Error(resp.Header().Code) == iproto.ER_UNKNOWN_REQUEST_TYPE {
+			// IPROTO_ID requests are not supported by server.
+			return info, nil
 		}
+		return info, err
 	}
 
 	if len(data) == 0 {
@@ -511,12 +506,12 @@ func readResponse(r io.Reader) (Response, error) {
 
 	respBytes, err := read(r, lenbuf[:])
 	if err != nil {
-		return &ConnResponse{}, fmt.Errorf("read error: %w", err)
+		return &BaseResponse{}, fmt.Errorf("read error: %w", err)
 	}
 
 	buf := smallBuf{b: respBytes}
 	header, err := decodeHeader(msgpack.NewDecoder(&smallBuf{}), &buf)
-	resp := &ConnResponse{header: header, buf: buf}
+	resp := &BaseResponse{header: header, buf: buf}
 	if err != nil {
 		return resp, fmt.Errorf("decode response header error: %w", err)
 	}