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, an optional argument `on_push_ctx`
is used in the form of a python list, where the
received data from out-of-band messages will be added.

Closes #201

Author: GRISHNOV

tarantool/connection.py

@@ -55,7 +55,8 @@
     REQUEST_TYPE_ERROR,
     IPROTO_GREETING_SIZE,
     ITERATOR_EQ,
-    ITERATOR_ALL
+    ITERATOR_ALL,
+    IPROTO_CHUNK
 )
 from tarantool.error import (
     Error,
@@ -748,7 +749,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, pushed_data=[], on_push=None, on_push_res=[]):
         """
         Send request without trying to reconnect.
         Reload schema, if required.
@@ -767,12 +768,30 @@ 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.
+                    pushed_data.append(response._data.copy())
+                    if callable(on_push):
+                        # Callback function with data from out-of-band 
+                        # message is being called.
+                        on_push_res.append(on_push(response._data.copy()))
+                    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 +870,7 @@ def check():  # Check that connection is alive
             self.wrap_socket_ssl()
         self.handshake()
 
-    def _send_request(self, request):
+    def _send_request(self, request, pushed_data=[], on_push=None, on_push_res=[]):
         """
         Send a request to the server through the socket.
 
@@ -872,7 +891,7 @@ def _send_request(self, request):
 
         self._opt_reconnect()
 
-        return self._send_request_wo_reconnect(request)
+        return self._send_request_wo_reconnect(request, pushed_data, on_push, on_push_res)
 
     def load_schema(self):
         """
@@ -914,7 +933,7 @@ def flush_schema(self):
         self.schema.flush()
         self.load_schema()
 
-    def call(self, func_name, *args):
+    def call(self, func_name, *args, **kwargs):
         """
         Execute a CALL request: call a stored Lua function.
 
@@ -930,19 +949,30 @@ 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)
 
         # This allows to use a tuple or list as an argument
         if len(args) == 1 and isinstance(args[0], (list, tuple)):
             args = args[0]
+        # Case for absence of optional arg for accepting out-of-band msg data
+        if not 'pushed_data' in kwargs:
+            kwargs['pushed_data'] = []
+        if not 'on_push' in kwargs:
+            kwargs['on_push'] = None
+        if not 'on_push_res' in kwargs:
+            kwargs['on_push_res'] = []
 
         request = RequestCall(self, func_name, args, self.call_16)
-        response = self._send_request(request)
+        response = self._send_request(request, kwargs['pushed_data'], kwargs['on_push'], kwargs['on_push_res'])
         return response
 
-    def eval(self, expr, *args):
+    def eval(self, expr, *args, **kwargs):
         """
         Execute an EVAL request: evaluate a Lua expression.
 
@@ -966,12 +996,19 @@ def eval(self, expr, *args):
         # This allows to use a tuple or list as an argument
         if len(args) == 1 and isinstance(args[0], (list, tuple)):
             args = args[0]
+        # Case for absence of optional arg for accepting out-of-band msg data
+        if not 'pushed_data' in kwargs:
+            kwargs['pushed_data'] = []
+        if not 'on_push' in kwargs:
+            kwargs['on_push'] = None
+        if not 'on_push_res' in kwargs:
+            kwargs['on_push_res'] = []
 
         request = RequestEval(self, expr, args)
-        response = self._send_request(request)
+        response = self._send_request(request, kwargs['pushed_data'], kwargs['on_push'], kwargs['on_push_res'])
         return response
 
-    def replace(self, space_name, values):
+    def replace(self, space_name, values, **kwargs):
         """
         Execute a REPLACE request: `replace`_ a tuple in the space.
         Doesn't throw an error if there is no tuple with the specified
@@ -994,10 +1031,18 @@ def replace(self, space_name, values):
         .. _replace: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/replace/
         """
 
+        # Case for absence of optional arg for accepting out-of-band msg data
+        if not 'pushed_data' in kwargs:
+            kwargs['pushed_data'] = []
+        if not 'on_push' in kwargs:
+            kwargs['on_push'] = None
+        if not 'on_push_res' in kwargs:
+            kwargs['on_push_res'] = []
+
         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, kwargs['pushed_data'], kwargs['on_push'], kwargs['on_push_res'])
 
     def authenticate(self, user, password):
         """
@@ -1149,7 +1194,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, **kwargs):
         """
         Execute an INSERT request: `insert`_ a tuple to the space.
         Throws an error if there is already a tuple with the same
@@ -1172,12 +1217,20 @@ def insert(self, space_name, values):
         .. _insert: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/insert/
         """
 
+        # Case for absence of optional arg for accepting out-of-band msg data
+        if not 'pushed_data' in kwargs:
+            kwargs['pushed_data'] = []
+        if not 'on_push' in kwargs:
+            kwargs['on_push'] = None
+        if not 'on_push_res' in kwargs:
+            kwargs['on_push_res'] = []
+
         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, kwargs['pushed_data'], kwargs['on_push'], kwargs['on_push_res'])
 
-    def delete(self, space_name, key, *, index=0):
+    def delete(self, space_name, key, *, index=0, **kwargs):
         """
         Execute a DELETE request: `delete`_ a tuple in the space.
 
@@ -1202,15 +1255,23 @@ def delete(self, space_name, key, *, index=0):
         .. _delete: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/delete/
         """
 
+        # Case for absence of optional arg for accepting out-of-band msg data
+        if not 'pushed_data' in kwargs:
+            kwargs['pushed_data'] = []
+        if not 'on_push' in kwargs:
+            kwargs['on_push'] = None
+        if not 'on_push_res' in kwargs:
+            kwargs['on_push_res'] = []
+
         key = check_key(key)
         if isinstance(space_name, str):
             space_name = self.schema.get_space(space_name).sid
         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, kwargs['pushed_data'], kwargs['on_push'], kwargs['on_push_res'])
 
-    def upsert(self, space_name, tuple_value, op_list, *, index=0):
+    def upsert(self, space_name, tuple_value, op_list, *, index=0, **kwargs):
         """
         Execute an UPSERT request: `upsert`_ a tuple to the space.
 
@@ -1252,16 +1313,24 @@ def upsert(self, space_name, tuple_value, op_list, *, index=0):
         .. _upsert: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/upsert/
         """
 
+        # Case for absence of optional arg for accepting out-of-band msg data
+        if not 'pushed_data' in kwargs:
+            kwargs['pushed_data'] = []
+        if not 'on_push' in kwargs:
+            kwargs['on_push'] = None
+        if not 'on_push_res' in kwargs:
+            kwargs['on_push_res'] = []
+
         if isinstance(space_name, str):
             space_name = self.schema.get_space(space_name).sid
         if isinstance(index, str):
             index = self.schema.get_index(space_name, index).iid
         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, kwargs['pushed_data'], kwargs['on_push'], kwargs['on_push_res'])
 
-    def update(self, space_name, key, op_list, *, index=0):
+    def update(self, space_name, key, op_list, *, index=0, **kwargs):
         """
         Execute an UPDATE request: `update`_ a tuple in the space.
 
@@ -1331,14 +1400,22 @@ def update(self, space_name, key, op_list, *, index=0):
         .. _update: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/update/
         """
 
+        # Case for absence of optional arg for accepting out-of-band msg data
+        if not 'pushed_data' in kwargs:
+            kwargs['pushed_data'] = []
+        if not 'on_push' in kwargs:
+            kwargs['on_push'] = None
+        if not 'on_push_res' in kwargs:
+            kwargs['on_push_res'] = []
+
         key = check_key(key)
         if isinstance(space_name, str):
             space_name = self.schema.get_space(space_name).sid
         if isinstance(index, str):
             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, kwargs['pushed_data'], kwargs['on_push'], kwargs['on_push_res'])
 
     def ping(self, notime=False):
         """
@@ -1368,7 +1445,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, **kwargs):
         """
         Execute a SELECT request: `select`_ a tuple from the space.
 
@@ -1518,13 +1595,21 @@ def select(self, space_name, key=None, *, offset=0, limit=0xffffffff, index=0, i
         # tuples)
         key = check_key(key, select=True)
 
+        # Case for absence of optional arg for accepting out-of-band msg data
+        if not 'pushed_data' in kwargs:
+            kwargs['pushed_data'] = []
+        if not 'on_push' in kwargs:
+            kwargs['on_push'] = None
+        if not 'on_push_res' in kwargs:
+            kwargs['on_push_res'] = []
+
         if isinstance(space_name, str):
             space_name = self.schema.get_space(space_name).sid
         if isinstance(index, str):
             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, kwargs['pushed_data'], kwargs['on_push'], kwargs['on_push_res'])
         return response
 
     def space(self, space_name):

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

tarantool/mesh_connection.py

@@ -581,7 +581,7 @@ def _opt_refresh_instances(self):
             update_connection(self, addr)
             self._opt_reconnect()
 
-    def _send_request(self, request):
+    def _send_request(self, request, pushed_data=[], on_push=None, on_push_res=[]):
         """
         Send a request to a Tarantool server. If required, refresh
         addresses list before sending a request.