Add support for ACKs in join()/subscribe()

tarantool/tarantool#1147

Author: rtsisyk

tarantool/connection.py

@@ -21,6 +21,7 @@
 from tarantool.response import Response
 from tarantool.request import (
     Request,
+    RequestOK,
     RequestCall,
     RequestDelete,
     RequestEval,
@@ -52,7 +53,7 @@
     NetworkWarning)
 
 from .schema import Schema
-from .utils import check_key, greeting_decode
+from .utils import check_key, greeting_decode, version_id
 
 
 class Connection(object):
@@ -357,8 +358,13 @@ def authenticate(self, user, password):
 
     def join(self, server_uuid):
         request = RequestJoin(self, server_uuid)
+        sync = request._sync
         resp = self._send_request(request)
         while True:
+            if self.version_id >= version_id(1, 7, 0):
+                # Send acknowledgement
+                ack = RequestOK(self, sync)
+                self._socket.sendall(bytes(ack))
             yield resp
             if resp.code == REQUEST_TYPE_OK or resp.code >= REQUEST_TYPE_ERROR:
                 return
@@ -368,8 +374,13 @@ def join(self, server_uuid):
     def subscribe(self, cluster_uuid, server_uuid, vclock={}):
         # FIXME rudnyh: ^ 'vclock={}'? really? sure?
         request = RequestSubscribe(self, cluster_uuid, server_uuid, vclock)
+        sync = request._sync
         resp = self._send_request(request)
         while True:
+            if self.version_id >= version_id(1, 7, 0):
+                # Send acknowledgement
+                ack = RequestOK(self, sync)
+                self._socket.sendall(bytes(ack))
             yield resp
             if resp.code >= REQUEST_TYPE_ERROR:
                 return

tarantool/request.py

@@ -24,6 +24,7 @@
     IPROTO_CLUSTER_UUID,
     IPROTO_VCLOCK,
     IPROTO_EXPR,
+    REQUEST_TYPE_OK,
     REQUEST_TYPE_PING,
     REQUEST_TYPE_SELECT,
     REQUEST_TYPE_INSERT,
@@ -278,3 +279,16 @@ def __init__(self, conn, cluster_uuid, server_uuid, vclock):
             IPROTO_VCLOCK: vclock
         })
         self._bytes = self.header(len(request_body)) + request_body
+
+class RequestOK(Request):
+    '''
+    Represents OK acknowledgement
+    '''
+    request_type = REQUEST_TYPE_OK
+
+    # pylint: disable=W0231
+    def __init__(self, conn, sync):
+        super(RequestOK, self).__init__(conn)
+        header = msgpack.dumps({IPROTO_CODE: self.request_type,
+                                IPROTO_SYNC: sync})
+        self._bytes = msgpack.dumps(len(header)) + header