pushAPI: remove box.session.push usage

CHANGELOG.md

@@ -20,6 +20,11 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 * Now cases of `<-ctx.Done()` returns wrapped error provided by `ctx.Cause()`.
   Allows you compare it using `errors.Is/As` (#457).
 * Removed deprecated `pool` methods, related interfaces and tests are updated (#478).
+* Removed deprecated `box.session.push()` usage: Future.AppendPush()
+  and Future.GetIterator() methods, ResponseIterator and TimeoutResponseIterator types,
+  Future.pushes[] (#480).
+* Removed `code == iproto.IPROTO_CHUNK` checking in `Connection.reader` due to removing 
+  `box.session.push()` usage (#480)
 
 ### Fixed
 

MIGRATION.md

@@ -12,6 +12,8 @@ TODO
 * `box.New` returns an error instead of panic
 * Added `box.MustNew` wrapper for `box.New` without an error
 * Removed deprecated `pool` methods, related interfaces and tests are updated.
+* Removed `box.session.push()` usage: Future.AppendPush() and Future.GetIterator()
+  methods, ResponseIterator and TimeoutResponseIterator types, `Future.pushes[]`.
 
 ## Migration from v1.x.x to v2.x.x
 

connection.go

@@ -882,15 +882,7 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 			}
 			continue
 		} else if code == iproto.IPROTO_CHUNK {
-			if fut = conn.peekFuture(header.RequestId); fut != nil {
-				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)
-				}
-			}
+			// not implemented
 		} else {
 			if fut = conn.fetchFuture(header.RequestId); fut != nil {
 				if err := fut.SetResponse(header, &buf); err != nil {
@@ -1131,26 +1123,6 @@ func (conn *Connection) markDone(fut *Future) {
 	conn.decrementRequestCnt()
 }
 
-func (conn *Connection) peekFuture(reqid uint32) (fut *Future) {
-	shard := &conn.shard[reqid&(conn.opts.Concurrency-1)]
-	pos := (reqid / conn.opts.Concurrency) & (requestsMap - 1)
-	shard.rmut.Lock()
-	defer shard.rmut.Unlock()
-
-	if conn.opts.Timeout > 0 {
-		if fut = conn.getFutureImp(reqid, true); fut != nil {
-			pair := &shard.requests[pos]
-			*pair.last = fut
-			pair.last = &fut.next
-			fut.timeout = time.Since(epoch) + conn.opts.Timeout
-		}
-	} else {
-		fut = conn.getFutureImp(reqid, false)
-	}
-
-	return fut
-}
-
 func (conn *Connection) fetchFuture(reqid uint32) (fut *Future) {
 	shard := &conn.shard[reqid&(conn.opts.Concurrency-1)]
 	shard.rmut.Lock()

example_test.go

@@ -1102,41 +1102,6 @@ func ExampleErrorNo() {
 	// Success.
 }
 
-func ExampleFuture_GetIterator() {
-	conn := exampleConnect(dialer, opts)
-	defer conn.Close()
-
-	const timeout = 3 * time.Second
-	fut := conn.Do(tarantool.NewCallRequest("push_func").
-		Args([]interface{}{4}),
-	)
-
-	var it tarantool.ResponseIterator
-	for it = fut.GetIterator().WithTimeout(timeout); it.Next(); {
-		resp := it.Value()
-		data, _ := resp.Decode()
-		if it.IsPush() {
-			// It is a push message.
-			fmt.Printf("push message: %v\n", data[0])
-		} else if resp.Header().Error == tarantool.ErrorNo {
-			// It is a regular response.
-			fmt.Printf("response: %v", data[0])
-		} else {
-			fmt.Printf("an unexpected response code %d", resp.Header().Error)
-		}
-	}
-	if err := it.Err(); err != nil {
-		fmt.Printf("error in call of push_func is %v", err)
-		return
-	}
-	// Output:
-	// push message: 1
-	// push message: 2
-	// push message: 3
-	// push message: 4
-	// response: 4
-}
-
 func ExampleConnect() {
 	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
 	defer cancel()

future.go

@@ -13,7 +13,6 @@ type Future struct {
 	next      *Future
 	timeout   time.Duration
 	mutex     sync.Mutex
-	pushes    []Response
 	resp      Response
 	err       error
 	ready     chan struct{}
@@ -39,131 +38,15 @@ func (fut *Future) isDone() bool {
 	}
 }
 
-type asyncResponseIterator struct {
-	fut     *Future
-	timeout time.Duration
-	resp    Response
-	err     error
-	curPos  int
-	done    bool
-}
-
-func (it *asyncResponseIterator) Next() bool {
-	if it.done || it.err != nil {
-		it.resp = nil
-		return false
-	}
-
-	var last = false
-	var exit = false
-	for !exit {
-		// We try to read at least once.
-		it.fut.mutex.Lock()
-		it.resp = it.nextResponse()
-		it.err = it.fut.err
-		last = it.resp == it.fut.resp
-		it.fut.mutex.Unlock()
-
-		if it.timeout == 0 || it.resp != nil || it.err != nil {
-			break
-		}
-
-		select {
-		case <-it.fut.ready:
-		case <-time.After(it.timeout):
-			exit = true
-		}
-	}
-
-	if it.resp == nil {
-		return false
-	}
-
-	if last {
-		it.done = true
-	} else {
-		it.err = nil
-		it.curPos += 1
-	}
-
-	return true
-}
-
-func (it *asyncResponseIterator) Value() Response {
-	return it.resp
-}
-
-func (it *asyncResponseIterator) IsPush() bool {
-	return !it.done
-}
-
-func (it *asyncResponseIterator) Err() error {
-	return it.err
-}
-
-func (it *asyncResponseIterator) WithTimeout(timeout time.Duration) TimeoutResponseIterator {
-	it.timeout = timeout
-	return it
-}
-
-func (it *asyncResponseIterator) nextResponse() (resp Response) {
-	fut := it.fut
-	pushesLen := len(fut.pushes)
-
-	if it.curPos < pushesLen {
-		resp = fut.pushes[it.curPos]
-	} else if it.curPos == pushesLen {
-		resp = fut.resp
-	}
-
-	return resp
-}
-
-// PushResponse is used for push requests for the Future.
-type PushResponse struct {
-	baseResponse
-}
-
-func createPushResponse(header Header, body io.Reader) (Response, error) {
-	resp, err := createBaseResponse(header, body)
-	if err != nil {
-		return nil, err
-	}
-	return &PushResponse{resp}, nil
-}
-
 // NewFuture creates a new empty Future for a given Request.
 func NewFuture(req Request) (fut *Future) {
 	fut = &Future{}
 	fut.ready = make(chan struct{}, 1000000000)
 	fut.done = make(chan struct{})
-	fut.pushes = make([]Response, 0)
 	fut.req = req
 	return fut
 }
 
-// AppendPush appends the push response to the future.
-// Note: it works only before SetResponse() or SetError()
-//
-// Deprecated: the method will be removed in the next major version,
-// use Connector.NewWatcher() instead of box.session.push().
-func (fut *Future) AppendPush(header Header, body io.Reader) error {
-	fut.mutex.Lock()
-	defer fut.mutex.Unlock()
-
-	if fut.isDone() {
-		return nil
-	}
-	resp, err := createPushResponse(header, body)
-	if err != nil {
-		return err
-	}
-	fut.pushes = append(fut.pushes, resp)
-
-	fut.ready <- struct{}{}
-	return nil
-}
-
 // SetResponse sets a response for the future and finishes the future.
 func (fut *Future) SetResponse(header Header, body io.Reader) error {
 	fut.mutex.Lock()
@@ -235,24 +118,6 @@ func (fut *Future) GetTyped(result interface{}) error {
 	return fut.resp.DecodeTyped(result)
 }
 
-// GetIterator returns an iterator for iterating through push messages
-// and a response. Push messages and the response will contain deserialized
-// result in Data field as for the Get() function.
-//
-// # See also
-//
-//   - box.session.push():
-//     https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_session/push/
-//
-// Deprecated: the method will be removed in the next major version,
-// use Connector.NewWatcher() instead of box.session.push().
-func (fut *Future) GetIterator() (it TimeoutResponseIterator) {
-	futit := &asyncResponseIterator{
-		fut: fut,
-	}
-	return futit
-}
-
 var closedChan = make(chan struct{})
 
 func init() {