Skip to content

api: remove box.session.push usage #494

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Nov 13, 2025
Merged

api: remove box.session.push usage #494

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,10 @@ 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()` support: Future.AppendPush()
and Future.GetIterator() methods, ResponseIterator and TimeoutResponseIterator types,
Future.pushes[] (#480).
* `LogAppendPushFailed` replaced with `LogBoxSessionPushUnsupported` (#480)

### Fixed

Expand Down
2 changes: 2 additions & 0 deletions MIGRATION.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()` support: Future.AppendPush() and Future.GetIterator()
methods, ResponseIterator and TimeoutResponseIterator types.

## Migration from v1.x.x to v2.x.x

Expand Down
40 changes: 6 additions & 34 deletions connection.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,8 +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
// LogBoxSessionPushUnsupported is logged when response type turned IPROTO_CHUNK.
LogBoxSessionPushUnsupported
)

// ConnEvent is sent throw Notify channel specified in Opts.
Expand Down Expand Up @@ -118,9 +118,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)
case LogBoxSessionPushUnsupported:
header := v[0].(Header)
log.Printf("tarantool: unsupported box.session.push() for request %d", header.RequestId)
default:
args := append([]interface{}{"tarantool: unexpected event ", event, conn}, v...)
log.Print(args...)
Expand Down Expand Up @@ -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)
}
}
conn.opts.Logger.Report(LogBoxSessionPushUnsupported, conn, header)
} else {
if fut = conn.fetchFuture(header.RequestId); fut != nil {
if err := fut.SetResponse(header, &buf); err != nil {
Expand Down Expand Up @@ -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()
Expand Down
35 changes: 0 additions & 35 deletions example_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
135 changes: 0 additions & 135 deletions future.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@ type Future struct {
next *Future
timeout time.Duration
mutex sync.Mutex
pushes []Response
resp Response
err error
ready chan struct{}
Expand All @@ -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()
Expand Down Expand Up @@ -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() {
Expand Down
Loading
Loading