iproto: support feature push

Adds support for receiving out-of-band messages
from a server that uses box.session.push call.

Data obtaining is possible for methods:
`call`, `eval`, `select`, `insert`, `replace`,
`update`, `upsert`, `delete`.
To do this, optional arguments `on_push` and `on_push_ctx`
are used for this methods.

Closes #201

Author: GRISHNOV

tarantool/connection.py

@@ -7,6 +7,7 @@
 import time
 import errno
 import socket
+import copy
 try:
     import ssl
     is_ssl_supported = True
@@ -55,7 +56,8 @@
     REQUEST_TYPE_ERROR,
     IPROTO_GREETING_SIZE,
     ITERATOR_EQ,
-    ITERATOR_ALL
+    ITERATOR_ALL,
+    IPROTO_CHUNK
 )
 from tarantool.error import (
     Error,
@@ -151,55 +153,56 @@ def connect(self):
         raise NotImplementedError
 
     @abc.abstractmethod
-    def call(self, func_name, *args):
+    def call(self, func_name, *args, on_push=None, on_push_ctx=None):
         """
         Reference implementation: :meth:`~tarantool.Connection.call`.
         """
 
         raise NotImplementedError
 
     @abc.abstractmethod
-    def eval(self, expr, *args):
+    def eval(self, expr, *args, on_push=None, on_push_ctx=None):
         """
         Reference implementation: :meth:`~tarantool.Connection.eval`.
         """
 
         raise NotImplementedError
 
     @abc.abstractmethod
-    def replace(self, space_name, values):
+    def replace(self, space_name, values, on_push=None, on_push_ctx=None):
         """
         Reference implementation: :meth:`~tarantool.Connection.replace`.
         """
 
         raise NotImplementedError
 
     @abc.abstractmethod
-    def insert(self, space_name, values):
+    def insert(self, space_name, values, on_push=None, on_push_ctx=None):
         """
         Reference implementation: :meth:`~tarantool.Connection.insert`.
         """
 
         raise NotImplementedError
 
     @abc.abstractmethod
-    def delete(self, space_name, key, *, index=None):
+    def delete(self, space_name, key, *, index=None, on_push=None, on_push_ctx=None):
         """
         Reference implementation: :meth:`~tarantool.Connection.delete`.
         """
 
         raise NotImplementedError
 
     @abc.abstractmethod
-    def upsert(self, space_name, tuple_value, op_list, *, index=None):
+    def upsert(self, space_name, tuple_value, op_list, *, index=None, 
+               on_push=None, on_push_ctx=None):
         """
         Reference implementation: :meth:`~tarantool.Connection.upsert`.
         """
 
         raise NotImplementedError
 
     @abc.abstractmethod
-    def update(self, space_name, key, op_list, *, index=None):
+    def update(self, space_name, key, op_list, *, index=None, on_push=None, on_push_ctx=None):
         """
         Reference implementation: :meth:`~tarantool.Connection.update`.
         """
@@ -216,7 +219,7 @@ def ping(self, notime):
 
     @abc.abstractmethod
     def select(self, space_name, key, *, offset=None, limit=None,
-               index=None, iterator=None):
+               index=None, iterator=None, on_push=None, on_push_ctx=None):
         """
         Reference implementation: :meth:`~tarantool.Connection.select`.
         """
@@ -748,7 +751,7 @@ def _read_response(self):
         # Read the packet
         return self._recv(length)
 
-    def _send_request_wo_reconnect(self, request):
+    def _send_request_wo_reconnect(self, request, on_push=None, on_push_ctx=None):
         """
         Send request without trying to reconnect.
         Reload schema, if required.
@@ -767,12 +770,33 @@ def _send_request_wo_reconnect(self, request):
 
         assert isinstance(request, Request)
 
+        # Flag for detecting the last message as out-of-band.
+        iproto_chunk_detected = False
+        
         response = None
         while True:
             try:
-                self._socket.sendall(bytes(request))
+                if not iproto_chunk_detected:
+                    # If the last received message is out-of-band, 
+                    # the request will not be sent.
+                    self._socket.sendall(bytes(request))
                 response = request.response_class(self, self._read_response())
-                break
+                if response._code == IPROTO_CHUNK:
+                    # Сase of receiving an out-of-band message.
+                    if on_push is not None: 
+                        if callable(on_push):
+                            # Callback function with data from out-of-band 
+                            # message is being called.
+                            # The callback result can be written to on_push_ctx.
+                            on_push(response._data, on_push_ctx)
+                        else:
+                            raise TypeError('The on_push callback must be callable')
+                    iproto_chunk_detected = True
+                    # Receiving the next message.
+                    continue
+                else:
+                    # Сase of receiving main message.
+                    break
             except SchemaReloadException as e:
                 self.update_schema(e.schema_version)
                 continue
@@ -851,7 +875,7 @@ def check():  # Check that connection is alive
             self.wrap_socket_ssl()
         self.handshake()
 
-    def _send_request(self, request):
+    def _send_request(self, request, on_push=None, on_push_ctx=None):
         """
         Send a request to the server through the socket.
 
@@ -872,7 +896,7 @@ def _send_request(self, request):
 
         self._opt_reconnect()
 
-        return self._send_request_wo_reconnect(request)
+        return self._send_request_wo_reconnect(request, on_push, on_push_ctx)
 
     def load_schema(self):
         """
@@ -914,7 +938,7 @@ def flush_schema(self):
         self.schema.flush()
         self.load_schema()
 
-    def call(self, func_name, *args):
+    def call(self, func_name, *args, on_push=None, on_push_ctx=None):
         """
         Execute a CALL request: call a stored Lua function.
 
@@ -930,6 +954,10 @@ def call(self, func_name, *args):
             :exc:`~tarantool.error.SchemaError`,
             :exc:`~tarantool.error.NetworkError`,
             :exc:`~tarantool.error.SslError`
+
+        !!!
+        TODO: write docs
+        !!!
         """
 
         assert isinstance(func_name, str)
@@ -939,10 +967,10 @@ def call(self, func_name, *args):
             args = args[0]
 
         request = RequestCall(self, func_name, args, self.call_16)
-        response = self._send_request(request)
+        response = self._send_request(request, on_push, on_push_ctx)
         return response
 
-    def eval(self, expr, *args):
+    def eval(self, expr, *args, on_push=None, on_push_ctx=None):
         """
         Execute an EVAL request: evaluate a Lua expression.
 
@@ -968,10 +996,10 @@ def eval(self, expr, *args):
             args = args[0]
 
         request = RequestEval(self, expr, args)
-        response = self._send_request(request)
+        response = self._send_request(request, on_push, on_push_ctx)
         return response
 
-    def replace(self, space_name, values):
+    def replace(self, space_name, values, on_push=None, on_push_ctx=None):
         """
         Execute a REPLACE request: `replace`_ a tuple in the space.
         Doesn't throw an error if there is no tuple with the specified
@@ -997,7 +1025,7 @@ def replace(self, space_name, values):
         if isinstance(space_name, str):
             space_name = self.schema.get_space(space_name).sid
         request = RequestReplace(self, space_name, values)
-        return self._send_request(request)
+        return self._send_request(request, on_push, on_push_ctx)
 
     def authenticate(self, user, password):
         """
@@ -1149,7 +1177,7 @@ def subscribe(self, cluster_uuid, server_uuid, vclock=None):
                 return
         self.close()  # close connection after SUBSCRIBE
 
-    def insert(self, space_name, values):
+    def insert(self, space_name, values, on_push=None, on_push_ctx=None):
         """
         Execute an INSERT request: `insert`_ a tuple to the space.
         Throws an error if there is already a tuple with the same
@@ -1175,9 +1203,9 @@ def insert(self, space_name, values):
         if isinstance(space_name, str):
             space_name = self.schema.get_space(space_name).sid
         request = RequestInsert(self, space_name, values)
-        return self._send_request(request)
+        return self._send_request(request, on_push, on_push_ctx)
 
-    def delete(self, space_name, key, *, index=0):
+    def delete(self, space_name, key, *, index=0, on_push=None, on_push_ctx=None):
         """
         Execute a DELETE request: `delete`_ a tuple in the space.
 
@@ -1208,9 +1236,9 @@ def delete(self, space_name, key, *, index=0):
         if isinstance(index, str):
             index = self.schema.get_index(space_name, index).iid
         request = RequestDelete(self, space_name, index, key)
-        return self._send_request(request)
+        return self._send_request(request, on_push, on_push_ctx)
 
-    def upsert(self, space_name, tuple_value, op_list, *, index=0):
+    def upsert(self, space_name, tuple_value, op_list, *, index=0, on_push=None, on_push_ctx=None):
         """
         Execute an UPSERT request: `upsert`_ a tuple to the space.
 
@@ -1259,9 +1287,9 @@ def upsert(self, space_name, tuple_value, op_list, *, index=0):
         op_list = self._ops_process(space_name, op_list)
         request = RequestUpsert(self, space_name, index, tuple_value,
                                 op_list)
-        return self._send_request(request)
+        return self._send_request(request, on_push, on_push_ctx)
 
-    def update(self, space_name, key, op_list, *, index=0):
+    def update(self, space_name, key, op_list, *, index=0, on_push=None, on_push_ctx=None):
         """
         Execute an UPDATE request: `update`_ a tuple in the space.
 
@@ -1338,7 +1366,7 @@ def update(self, space_name, key, op_list, *, index=0):
             index = self.schema.get_index(space_name, index).iid
         op_list = self._ops_process(space_name, op_list)
         request = RequestUpdate(self, space_name, index, key, op_list)
-        return self._send_request(request)
+        return self._send_request(request, on_push, on_push_ctx)
 
     def ping(self, notime=False):
         """
@@ -1368,7 +1396,7 @@ def ping(self, notime=False):
             return "Success"
         return t1 - t0
 
-    def select(self, space_name, key=None, *, offset=0, limit=0xffffffff, index=0, iterator=None):
+    def select(self, space_name, key=None, *, offset=0, limit=0xffffffff, index=0, iterator=None, on_push=None, on_push_ctx=None):
         """
         Execute a SELECT request: `select`_ a tuple from the space.
 
@@ -1524,7 +1552,7 @@ def select(self, space_name, key=None, *, offset=0, limit=0xffffffff, index=0, i
             index = self.schema.get_index(space_name, index).iid
         request = RequestSelect(self, space_name, index, key, offset,
                                 limit, iterator)
-        response = self._send_request(request)
+        response = self._send_request(request, on_push, on_push_ctx)
         return response
 
     def space(self, space_name):

tarantool/connection_pool.py

@@ -711,7 +711,7 @@ def _send(self, mode, method_name, *args, **kwargs):
 
         return resp
 
-    def call(self, func_name, *args, mode=None):
+    def call(self, func_name, *args, mode=None, on_push=None, on_push_ctx=None):
         """
         Execute a CALL request on the pool server: call a stored Lua
         function. Refer to :meth:`~tarantool.Connection.call`.
@@ -734,9 +734,9 @@ def call(self, func_name, *args, mode=None):
         if mode is None:
             raise ValueError("Please, specify 'mode' keyword argument")
 
-        return self._send(mode, 'call', func_name, *args)
+        return self._send(mode, 'call', func_name, *args, on_push=on_push, on_push_ctx=on_push_ctx)
 
-    def eval(self, expr, *args, mode=None):
+    def eval(self, expr, *args, mode=None, on_push=None, on_push_ctx=None):
         """
         Execute an EVAL request on the pool server: evaluate a Lua
         expression. Refer to :meth:`~tarantool.Connection.eval`.
@@ -759,9 +759,9 @@ def eval(self, expr, *args, mode=None):
         if mode is None:
             raise ValueError("Please, specify 'mode' keyword argument")
 
-        return self._send(mode, 'eval', expr, *args)
+        return self._send(mode, 'eval', expr, *args, on_push=on_push, on_push_ctx=on_push_ctx)
 
-    def replace(self, space_name, values, *, mode=Mode.RW):
+    def replace(self, space_name, values, *, mode=Mode.RW, on_push=None, on_push_ctx=None):
         """
         Execute a REPLACE request on the pool server: `replace`_ a tuple
         in the space. Refer to :meth:`~tarantool.Connection.replace`.
@@ -782,9 +782,9 @@ def replace(self, space_name, values, *, mode=Mode.RW):
         .. _replace: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/replace/
         """
 
-        return self._send(mode, 'replace', space_name, values)
+        return self._send(mode, 'replace', space_name, values, on_push=on_push, on_push_ctx=on_push_ctx)
 
-    def insert(self, space_name, values, *, mode=Mode.RW):
+    def insert(self, space_name, values, *, mode=Mode.RW, on_push=None, on_push_ctx=None):
         """
         Execute an INSERT request on the pool server: `insert`_ a tuple
         to the space. Refer to :meth:`~tarantool.Connection.insert`.
@@ -805,9 +805,9 @@ def insert(self, space_name, values, *, mode=Mode.RW):
         .. _insert: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/insert/
         """
 
-        return self._send(mode, 'insert', space_name, values)
+        return self._send(mode, 'insert', space_name, values, on_push=on_push, on_push_ctx=on_push_ctx)
 
-    def delete(self, space_name, key, *, index=0, mode=Mode.RW):
+    def delete(self, space_name, key, *, index=0, mode=Mode.RW, on_push=None, on_push_ctx=None):
         """
         Execute an DELETE request on the pool server: `delete`_ a tuple
         in the space. Refer to :meth:`~tarantool.Connection.delete`.
@@ -831,9 +831,9 @@ def delete(self, space_name, key, *, index=0, mode=Mode.RW):
         .. _delete: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/delete/
         """
 
-        return self._send(mode, 'delete', space_name, key, index=index)
+        return self._send(mode, 'delete', space_name, key, index=index, on_push=on_push, on_push_ctx=on_push_ctx)
 
-    def upsert(self, space_name, tuple_value, op_list, *, index=0, mode=Mode.RW):
+    def upsert(self, space_name, tuple_value, op_list, *, index=0, mode=Mode.RW, on_push=None, on_push_ctx=None):
         """
         Execute an UPSERT request on the pool server: `upsert`_ a tuple to
         the space. Refer to :meth:`~tarantool.Connection.upsert`.
@@ -861,9 +861,9 @@ def upsert(self, space_name, tuple_value, op_list, *, index=0, mode=Mode.RW):
         """
 
         return self._send(mode, 'upsert', space_name, tuple_value,
-            op_list, index=index)
+            op_list, index=index, on_push=on_push, on_push_ctx=on_push_ctx)
 
-    def update(self, space_name, key, op_list, *, index=0, mode=Mode.RW):
+    def update(self, space_name, key, op_list, *, index=0, mode=Mode.RW, on_push=None, on_push_ctx=None):
         """
         Execute an UPDATE request on the pool server: `update`_ a tuple
         in the space. Refer to :meth:`~tarantool.Connection.update`.
@@ -891,7 +891,7 @@ def update(self, space_name, key, op_list, *, index=0, mode=Mode.RW):
         """
 
         return self._send(mode, 'update', space_name, key, 
-            op_list, index=index)
+            op_list, index=index, on_push=on_push, on_push_ctx=on_push_ctx)
 
     def ping(self, notime=False, *, mode=None):
         """
@@ -917,7 +917,7 @@ def ping(self, notime=False, *, mode=None):
         return self._send(mode, 'ping', notime)
 
     def select(self, space_name, key, *, offset=0, limit=0xffffffff,
-               index=0, iterator=None, mode=Mode.ANY):
+               index=0, iterator=None, mode=Mode.ANY, on_push=None, on_push_ctx=None):
         """
         Execute a SELECT request on the pool server: `update`_ a tuple
         from the space. Refer to :meth:`~tarantool.Connection.select`.
@@ -951,7 +951,7 @@ def select(self, space_name, key, *, offset=0, limit=0xffffffff,
         """
 
         return self._send(mode, 'select', space_name, key, offset=offset, limit=limit,
-                          index=index, iterator=iterator)
+                          index=index, iterator=iterator, on_push=on_push, on_push_ctx=on_push_ctx)
 
     def execute(self, query, params=None, *, mode=None):
         """

tarantool/const.py

@@ -38,6 +38,7 @@
 
 IPROTO_GREETING_SIZE = 128
 IPROTO_BODY_MAX_LEN = 2147483648
+IPROTO_CHUNK=0x80
 
 REQUEST_TYPE_OK = 0
 REQUEST_TYPE_SELECT = 1