Add basiс support for JOIN and SUBSCRIBE commands

Author: rtsisyk

tarantool/connection.py

@@ -27,9 +27,11 @@
     RequestDelete,
     RequestEval,
     RequestInsert,
+    RequestJoin,
     RequestReplace,
     RequestPing,
     RequestSelect,
+    RequestSubscribe,
     RequestUpdate,
     RequestAuthenticate)
 
@@ -39,6 +41,8 @@
     RECONNECT_MAX_ATTEMPTS,
     RECONNECT_DELAY,
     RETRY_MAX_ATTEMPTS,
+    REQUEST_TYPE_OK,
+    REQUEST_TYPE_ERROR,
     IPROTO_GREETING_SIZE)
 from tarantool.error import (
     NetworkError,
@@ -332,6 +336,27 @@ def authenticate(self, user, password):
                                       self.password)
         return self._send_request_wo_reconnect(request)
 
+    def join(self, server_uuid):
+        request = RequestJoin(self, server_uuid)
+        resp = self._send_request(request)
+        while True:
+            yield resp
+            if resp.code == REQUEST_TYPE_OK or \
+               resp.code >= REQUEST_TYPE_ERROR:
+                return
+            resp = Response(self, self._read_response())
+        self.close() # close connection after JOIN
+
+    def subscribe(self, cluster_uuid, server_uuid, vclock = {}):
+        request = RequestSubscribe(self, cluster_uuid, server_uuid, vclock)
+        resp = self._send_request(request)
+        while True:
+            yield resp
+            if resp.code >= REQUEST_TYPE_ERROR:
+                return
+            resp = Response(self, self._read_response())
+        self.close() # close connection after SUBSCRIBE
+
     def insert(self, space_name, values):
         '''
         Execute INSERT request.

tarantool/const.py

@@ -12,14 +12,16 @@
 IPROTO_TUPLE = 0x21
 IPROTO_FUNCTION_NAME = 0x22
 IPROTO_USER_NAME = 0x23
+IPROTO_SERVER_UUID = 0x24
+IPROTO_CLUSTER_UUID = 0x25
+IPROTO_VCLOCK = 0x26
 IPROTO_EXPR = 0x27
 IPROTO_DATA = 0x30
 IPROTO_ERROR = 0x31
 
 IPROTO_GREETING_SIZE = 128
 
 REQUEST_TYPE_OK = 0
-REQUEST_TYPE_PING = 64
 REQUEST_TYPE_SELECT = 1
 REQUEST_TYPE_INSERT = 2
 REQUEST_TYPE_REPLACE = 3
@@ -28,6 +30,9 @@
 REQUEST_TYPE_CALL = 6
 REQUEST_TYPE_AUTHENTICATE = 7
 REQUEST_TYPE_EVAL = 8
+REQUEST_TYPE_PING = 64
+REQUEST_TYPE_JOIN = 65
+REQUEST_TYPE_SUBSCRIBE = 66
 REQUEST_TYPE_ERROR = 1 << 15
 
 

tarantool/request.py

@@ -19,6 +19,9 @@
     IPROTO_TUPLE,
     IPROTO_FUNCTION_NAME,
     IPROTO_ITERATOR,
+    IPROTO_SERVER_UUID,
+    IPROTO_CLUSTER_UUID,
+    IPROTO_VCLOCK,
     IPROTO_EXPR,
     REQUEST_TYPE_PING,
     REQUEST_TYPE_SELECT,
@@ -28,7 +31,9 @@
     REQUEST_TYPE_UPDATE,
     REQUEST_TYPE_CALL,
     REQUEST_TYPE_EVAL,
-    REQUEST_TYPE_AUTHENTICATE
+    REQUEST_TYPE_AUTHENTICATE,
+    REQUEST_TYPE_JOIN,
+    REQUEST_TYPE_SUBSCRIBE
 )
 
 
@@ -226,3 +231,35 @@ class RequestPing(Request):
     def __init__(self, conn):
         super(RequestPing, self).__init__(conn)
         self._bytes = self.header(0)
+
+class RequestJoin(Request):
+
+    '''
+        Represents JOIN request
+    '''
+    request_type = REQUEST_TYPE_JOIN
+
+    # pylint: disable=W0231
+    def __init__(self, conn, server_uuid):
+        super(RequestJoin, self).__init__(conn)
+        request_body = msgpack.dumps({ IPROTO_SERVER_UUID: server_uuid })
+        self._bytes = self.header(len(request_body)) + request_body
+
+class RequestSubscribe(Request):
+
+    '''
+        Represents SUBSCRIBE request
+    '''
+    request_type = REQUEST_TYPE_SUBSCRIBE
+
+    # pylint: disable=W0231
+    def __init__(self, conn, cluster_uuid, server_uuid, vclock):
+        super(RequestSubscribe, self).__init__(conn)
+        assert isinstance(vclock, dict)
+
+        request_body = msgpack.dumps({
+            IPROTO_CLUSTER_UUID: cluster_uuid,
+            IPROTO_SERVER_UUID: server_uuid,
+            IPROTO_VCLOCK: vclock
+        })
+        self._bytes = self.header(len(request_body)) + request_body

tarantool/response.py

@@ -48,26 +48,26 @@ def __init__(self, conn, response):
 
         self._sync = header.get(IPROTO_SYNC, 0)
         self.conn = conn
-        code = header[IPROTO_CODE]
-        body = None
+        self._code = header[IPROTO_CODE]
+        self._body = {}
         try:
-            body = unpacker.unpack()
+            self._body = unpacker.unpack()
         except msgpack.OutOfData:
-            body = {}
+            pass
 
-        if code == REQUEST_TYPE_OK:
+        if self._code < REQUEST_TYPE_ERROR:
             self._return_code = 0;
             self._completion_status = 0
-            self._data = body.get(IPROTO_DATA, None)
+            self._data = self._body.get(IPROTO_DATA, None)
             # Backward-compatibility
             if isinstance(self._data, (list, tuple)):
                 self.extend(self._data)
             else:
                 self.append(self._data)
         else:
             # Separate return_code and completion_code
-            self._return_message = body.get(IPROTO_ERROR, "")
-            self._return_code = code & (REQUEST_TYPE_ERROR - 1)
+            self._return_message = self._body.get(IPROTO_ERROR, "")
+            self._return_code = self._code & (REQUEST_TYPE_ERROR - 1)
             self._completion_status = 2
             self._data = None
             if self.conn.error:
@@ -99,6 +99,26 @@ def rowcount(self):
         '''
         return len(self)
 
+    @property
+    def body(self):
+        '''\
+        :type: dict
+
+        Required field in the server response.
+        Contains raw response body.
+        '''
+        return self._body
+
+    @property
+    def code(self):
+        '''\
+        :type: int
+
+        Required field in the server response.
+        Contains response type id.
+        '''
+        return self._code
+
     @property
     def return_code(self):
         '''\