api: add ability to mock connections in tests

Create a mock implementations `MockRequest`, `MockResponse` and
`MockDoer`.
The last one allows to mock not the full `Connection`, but its
part -- a structure, that implements new `Doer` interface
(a `Do` function).

Also added missing comments, all the changes are recorded in
the `CHANGELOG` and `README` files. Added new tests and
examples.

So this entity is easier to implement and it is enough to
mock tests that require working `Connection`.
All new mock structs and an example for `MockDoer` usage
are added to the `test_helpers` package.

Added new structs `MockDoer`, `MockRequest` to `test_helpers`.
Renamed `StrangerResponse` to `MockResponse`.

Added new connection log constant: `LogAppendPushFailed`.
It is logged when connection fails to append a push response.

Closes #237

Author: DerekBum

CHANGELOG.md

@@ -28,6 +28,13 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
   in requests instead of their IDs.
 - `GetSchema` function to get the actual schema (#7)
 - Support connection via an existing socket fd (#321)
+- Ability to mock connections for tests (#237). Added new types `MockDoer`, 
+  `MockRequest` to `test_helpers`.
+- `Response` method added to the `Request` interface (#237)
+- `Header` struct for the response header (#237). It can be accessed via 
+  `Header()` method of the `Response` interface.
+- New `LogAppendPushFailed` connection log constant (#237).
+  It is logged when connection fails to append a push response.
 
 ### Changed
 
@@ -67,6 +74,21 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
   it (#321)
 - Rename `pool.GetPoolInfo` to `pool.GetInfo`. Change return type to
   `map[string]ConnectionInfo` (#321)
+- All responses are now implementations of an `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.
+- `Future` constructors now accept `Request` as their argument (#237). 
+  Method `Get` now returns response data. To get the actual response new method 
+  added `GetResponse`. Methods `AppendPush` and `SetResponse` accepts 
+  response `Header` and data as their arguments.
+- All deprecated asynchronous requests operations on a `Connector` return 
+  response data instead of an actual responses (#237)
+- Renamed `StrangerResponse` to `MockResponse` (#237)
 
 ### Deprecated
 

README.md

@@ -128,12 +128,13 @@ func main() {
 	if err != nil {
 		fmt.Println("Connection refused:", err)
 	}
-	resp, err := conn.Do(tarantool.NewInsertRequest(999).
+	data, err := conn.Do(tarantool.NewInsertRequest(999).
 		Tuple([]interface{}{99999, "BB"}),
 	).Get()
 	if err != nil {
 		fmt.Println("Error", err)
-		fmt.Println("Code", resp.Code)
+	} else {
+		fmt.Printf("Data: %v", data)
 	}
 }
 ```
@@ -201,12 +202,18 @@ 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`.
+* All deprecated asynchronous requests operations on a `Pooler` return response 
+  data instead of an actual responses.
 
 #### crud package
 
 * `crud` operations `Timeout` option has `crud.OptFloat64` type
   instead of `crud.OptUint`.
 
+#### test_helpers package
+
+Renamed `StrangerResponse` to `MockResponse`.
+
 #### msgpack.v5
 
 Most function names and argument types in `msgpack.v5` and `msgpack.v2`
@@ -241,7 +248,10 @@ of the requests is an array instead of array of arrays.
 
 #### IPROTO constants
 
-IPROTO constants have been moved to a separate package [go-iproto](https://github.com/tarantool/go-iproto).
+* IPROTO constants have been moved to a separate package [go-iproto](https://github.com/tarantool/go-iproto).
+* `PushCode` constant is removed. To get information, if the current response is 
+  a push response, new `IsPush()` method of the response iterator could be used 
+  instead.
 
 #### Request changes
 
@@ -254,6 +264,34 @@ 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
+
+* New interface `Response` added.
+* 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.
+* Response header stored in a new `Header` struct. It could be accessed via 
+  `Header()` method.
+* Method `IsPush()` is added to the `ResponseIterator` interface.
+  It returns true if the current response is a push response.
+
+#### Future changes
+
+* `Future` constructors now accept `Request` as their argument.
+* Method `Get` now returns response data instead of the actual response.
+* New method `GetResponse` added to get an actual response.
+* Methods `AppendPush` and `SetResponse` accepts response `Header` and data 
+  as their arguments.
+
+#### Connector changes
+
+* All deprecated asynchronous requests operations on a `Connector` return 
+  response data instead of an actual responses.
+* New interface `Doer` is added as a child-interface instead of a `Do` method.
 
 #### Connect function
 
@@ -270,7 +308,7 @@ for creating a connection are now stored in corresponding `Dialer`, not in `Opts
 #### Connection schema
 
 * Removed `Schema` field from the `Connection` struct. Instead, new
-`GetSchema(Connector)` function was added to get the actual connection
+`GetSchema(Doer)` function was added to get the actual connection
 schema on demand.
 * `OverrideSchema(*Schema)` method replaced with the `SetSchema(Schema)`.
 

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...)
@@ -808,7 +813,7 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 			return
 		}
 		buf := smallBuf{b: respBytes}
-		header, err := decodeHeader(conn.dec, &buf)
+		header, code, err := decodeHeader(conn.dec, &buf)
 		if err != nil {
 			err = ClientError{
 				ErrProtocolError,
@@ -819,7 +824,7 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 		}
 
 		var fut *Future = nil
-		if iproto.Type(header.Code) == iproto.IPROTO_EVENT {
+		if code == iproto.IPROTO_EVENT {
 			if event, err := readWatchEvent(&buf); err == nil {
 				events <- event
 			} else {
@@ -830,13 +835,21 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 				conn.opts.Logger.Report(LogWatchEventReadFailed, conn, err)
 			}
 			continue
-		} else if header.Code == PushCode {
+		} else if code == iproto.IPROTO_CHUNK {
 			if fut = conn.peekFuture(header.RequestId); fut != nil {
-				fut.AppendPush(header, &buf)
+				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(header, &buf)
+				if err := fut.SetResponse(header, &buf); err != nil {
+					fut.SetError(fmt.Errorf("failed to set response: %w", err))
+				}
 				conn.markDone(fut)
 			}
 		}
@@ -874,8 +887,7 @@ func (conn *Connection) eventer(events <-chan connWatchEvent) {
 
 func (conn *Connection) newFuture(req Request) (fut *Future) {
 	ctx := req.Ctx()
-	fut = NewFuture()
-	fut.SetRequest(req)
+	fut = NewFuture(req)
 	if conn.rlimit != nil && conn.opts.RLimitAction == RLimitDrop {
 		select {
 		case conn.rlimit <- struct{}{}:
@@ -1056,7 +1068,7 @@ func (conn *Connection) putFuture(fut *Future, req Request, streamId uint64) {
 		if fut = conn.fetchFuture(reqid); fut != nil {
 			header := Header{
 				RequestId: reqid,
-				Code:      OkCode,
+				Error:     ErrorNo,
 			}
 			fut.SetResponse(header, nil)
 			conn.markDone(fut)
@@ -1204,7 +1216,7 @@ func (conn *Connection) nextRequestId(context bool) (requestId uint32) {
 func (conn *Connection) Do(req Request) *Future {
 	if connectedReq, ok := req.(ConnectedRequest); ok {
 		if connectedReq.Conn() != conn {
-			fut := NewFuture()
+			fut := NewFuture(req)
 			fut.SetError(errUnknownRequest)
 			return fut
 		}

connector.go

@@ -2,14 +2,20 @@ package tarantool
 
 import "time"
 
+// Doer is an interface that performs requests asynchronously.
+type Doer interface {
+	// Do performs a request asynchronously.
+	Do(req Request) (fut *Future)
+}
+
 type Connector interface {
+	Doer
 	ConnectedNow() bool
 	Close() error
 	ConfiguredTimeout() time.Duration
 	NewPrepared(expr string) (*Prepared, error)
 	NewStream() (*Stream, error)
 	NewWatcher(key string, callback WatchCallback) (Watcher, error)
-	Do(req Request) (fut *Future)
 
 	// Deprecated: the method will be removed in the next major version,
 	// use a PingRequest object + Do() instead.

const.go

@@ -9,6 +9,7 @@ const (
 )
 
 const (
-	OkCode   = uint32(iproto.IPROTO_OK)
-	PushCode = uint32(iproto.IPROTO_CHUNK)
+	// ErrorNo indicates whether an error is occurred. It could be used to
+	// check that a response has an error without the response body decoding.
+	ErrorNo = iproto.ER_UNKNOWN
 )

crud/select.go

@@ -2,7 +2,6 @@ package crud
 
 import (
 	"context"
-	"io"
 
 	"github.com/vmihailenco/msgpack/v5"
 
@@ -134,9 +133,3 @@ func (req SelectRequest) Context(ctx context.Context) SelectRequest {
 
 	return req
 }
-
-// Response creates a response for the SelectRequest.
-func (req SelectRequest) Response(header tarantool.Header,
-	body io.Reader) (tarantool.Response, error) {
-	return req.impl.Response(header, body)
-}

dial.go

@@ -398,16 +398,17 @@ func identify(w writeFlusher, r io.Reader) (ProtocolInfo, error) {
 		return info, err
 	}
 
-	resp, err := readResponse(r)
+	resp, err := readResponse(r, req)
 	if err != nil {
+		if resp != nil &&
+			resp.Header().Error == iproto.ER_UNKNOWN_REQUEST_TYPE {
+			// IPROTO_ID requests are not supported by server.
+			return info, nil
+		}
 		return info, err
 	}
 	data, err := resp.Decode()
 	if err != nil {
-		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
 	}
 
@@ -477,7 +478,7 @@ func authenticate(c Conn, auth Auth, user string, pass string, salt string) erro
 	if err = writeRequest(c, req); err != nil {
 		return err
 	}
-	if _, err = readResponse(c); err != nil {
+	if _, err = readResponse(c, req); err != nil {
 		return err
 	}
 	return nil
@@ -501,19 +502,31 @@ func writeRequest(w writeFlusher, req Request) error {
 }
 
 // readResponse reads a response from the reader.
-func readResponse(r io.Reader) (Response, error) {
+func readResponse(r io.Reader, req Request) (Response, error) {
 	var lenbuf [packetLengthBytes]byte
 
 	respBytes, err := read(r, lenbuf[:])
 	if err != nil {
-		return &BaseResponse{}, fmt.Errorf("read error: %w", err)
+		return nil, fmt.Errorf("read error: %w", err)
 	}
 
 	buf := smallBuf{b: respBytes}
-	header, err := decodeHeader(msgpack.NewDecoder(&smallBuf{}), &buf)
-	resp := &BaseResponse{header: header, buf: buf}
+	header, _, err := decodeHeader(msgpack.NewDecoder(&smallBuf{}), &buf)
 	if err != nil {
-		return resp, fmt.Errorf("decode response header error: %w", err)
+		return nil, fmt.Errorf("decode response header error: %w", err)
+	}
+	resp, err := req.Response(header, &buf)
+	if err != nil {
+		return nil, fmt.Errorf("creating response error: %w", err)
+	}
+	_, err = resp.Decode()
+	if err != nil {
+		switch err.(type) {
+		case Error:
+			return resp, err
+		default:
+			return resp, fmt.Errorf("decode response body error: %w", err)
+		}
 	}
 	return resp, nil
 }