api: remove join and subscribe

This is a breaking change.

Current join and subscribe implementations are rather useless. Connector
does not provide any API to process incoming replication requests. The
only supported scenario is to "connect as replica, skip everything that
has been sent through replication, close on error".

Current Tarantool team product strategy is to develop CDC features,
including replication support in language connectors, as Enterprise
edition products [1]. Since we don't plan to provide proper join and
subscribe implementation in the open-source connector in the future, we
decide to drop current half-baked implementation to not confuse new
users.

1. tarantool/go-tarantool#203

Author: DifferentialOrange

CHANGELOG.md

@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
     exceptions. `datetime.datetime` exceptions will
     be thrown instead of them.
   - Drop the support of `__eq__` operator for `pandas.Timestamp`.
+- **Breaking**: Remove `join` and `subscribe` connection methods.
 
 ## 0.12.1 - 2023-02-28
 

tarantool/connection.py

@@ -25,21 +25,17 @@
 
 from tarantool.response import (
     unpacker_factory as default_unpacker_factory,
-    Response,
 )
 from tarantool.request import (
     packer_factory as default_packer_factory,
     Request,
-    # RequestOK,
     RequestCall,
     RequestDelete,
     RequestEval,
     RequestInsert,
-    RequestJoin,
     RequestReplace,
     RequestPing,
     RequestSelect,
-    RequestSubscribe,
     RequestUpdate,
     RequestUpsert,
     RequestAuthenticate,
@@ -60,8 +56,6 @@
     DEFAULT_SSL_CIPHERS,
     DEFAULT_SSL_PASSWORD,
     DEFAULT_SSL_PASSWORD_FILE,
-    REQUEST_TYPE_OK,
-    REQUEST_TYPE_ERROR,
     IPROTO_GREETING_SIZE,
     ITERATOR_EQ,
     ITERATOR_ALL,
@@ -1520,61 +1514,6 @@ def _get_auth_type(self):
 
         return auth_type
 
-    def _join_v16(self, server_uuid):
-        """
-        Execute a JOIN request for Tarantool 1.6 and older.
-
-        :param server_uuid: UUID of Tarantool server to join.
-        :type server_uuid: :obj:`str`
-
-        :raise: :exc:`~AssertionError`,
-            :exc:`~tarantool.error.DatabaseError`,
-            :exc:`~tarantool.error.SchemaError`,
-            :exc:`~tarantool.error.NetworkError`,
-            :exc:`~tarantool.error.SslError`
-        """
-
-        request = RequestJoin(self, server_uuid)
-        self._socket.sendall(bytes(request))
-
-        while True:
-            resp = Response(self, self._read_response())
-            yield resp
-            if resp.code == REQUEST_TYPE_OK or resp.code >= REQUEST_TYPE_ERROR:
-                return
-        self.close()  # close connection after JOIN
-
-    def _join_v17(self, server_uuid):
-        """
-        Execute a JOIN request for Tarantool 1.7 and newer.
-
-        :param server_uuid: UUID of Tarantool server to join.
-        :type server_uuid: :obj:`str`
-
-        :raise: :exc:`~AssertionError`,
-            :exc:`~tarantool.error.DatabaseError`,
-            :exc:`~tarantool.error.SchemaError`,
-            :exc:`~tarantool.error.NetworkError`,
-            :exc:`~tarantool.error.SslError`
-        """
-
-        request = RequestJoin(self, server_uuid)
-        self._socket.sendall(bytes(request))
-        state = JoinState.HANDSHAKE
-        while True:
-            resp = Response(self, self._read_response())
-            yield resp
-            if resp.code >= REQUEST_TYPE_ERROR:
-                return
-            if resp.code == REQUEST_TYPE_OK:
-                if state == JoinState.HANDSHAKE:
-                    state = JoinState.INITIAL
-                elif state == JoinState.INITIAL:
-                    state = JoinState.FINAL
-                elif state == JoinState.FINAL:
-                    state = JoinState.DONE
-                    return
-
     def _ops_process(self, space, update_ops):
         new_ops = []
         for operation in update_ops:
@@ -1584,60 +1523,6 @@ def _ops_process(self, space, update_ops):
             new_ops.append(operation)
         return new_ops
 
-    def join(self, server_uuid):
-        """
-        Execute a JOIN request: `join`_ a replicaset.
-
-        :param server_uuid: UUID of connector "server".
-        :type server_uuid: :obj:`str`
-
-        :raise: :exc:`~AssertionError`,
-            :exc:`~tarantool.error.DatabaseError`,
-            :exc:`~tarantool.error.SchemaError`,
-            :exc:`~tarantool.error.NetworkError`,
-            :exc:`~tarantool.error.SslError`
-
-        .. _join: https://www.tarantool.io/en/doc/latest/dev_guide/internals/box_protocol/#iproto-join-0x41
-        """
-
-        self._opt_reconnect()
-        if self.version_id < version_id(1, 7, 0):
-            return self._join_v16(server_uuid)
-        return self._join_v17(server_uuid)
-
-    def subscribe(self, cluster_uuid, server_uuid, vclock=None):
-        """
-        Execute a SUBSCRIBE request: `subscribe`_ to a replicaset
-        updates. Connection is closed after subscribing.
-
-        :param cluster_uuid: UUID of replicaset cluster.
-        :type cluster_uuid: :obj:`str`
-
-        :param server_uuid: UUID of connector "server".
-        :type server_uuid: :obj:`str`
-
-        :param vclock: Connector "server" vclock.
-        :type vclock: :obj:`dict` or :obj:`None`, optional
-
-        :raise: :exc:`~AssertionError`,
-            :exc:`~tarantool.error.DatabaseError`,
-            :exc:`~tarantool.error.SchemaError`,
-            :exc:`~tarantool.error.NetworkError`,
-            :exc:`~tarantool.error.SslError`
-
-        .. _subscribe: https://www.tarantool.io/en/doc/latest/dev_guide/internals/box_protocol/#iproto-subscribe-0x42
-        """
-
-        vclock = vclock or {}
-        request = RequestSubscribe(self, cluster_uuid, server_uuid, vclock)
-        self._socket.sendall(bytes(request))
-        while True:
-            resp = Response(self, self._read_response())
-            yield resp
-            if resp.code >= REQUEST_TYPE_ERROR:
-                return
-        self.close()  # close connection after SUBSCRIBE
-
     def insert(self, space_name, values, *, on_push=None, on_push_ctx=None):
         """
         Execute an INSERT request: `insert`_ a tuple to the space.

tarantool/request.py

@@ -20,12 +20,8 @@
     IPROTO_TUPLE,
     IPROTO_FUNCTION_NAME,
     IPROTO_ITERATOR,
-    IPROTO_SERVER_UUID,
-    IPROTO_CLUSTER_UUID,
-    IPROTO_VCLOCK,
     IPROTO_EXPR,
     IPROTO_OPS,
-    # IPROTO_INDEX_BASE,
     IPROTO_SCHEMA_ID,
     IPROTO_SQL_TEXT,
     IPROTO_SQL_BIND,
@@ -44,8 +40,6 @@
     REQUEST_TYPE_EXECUTE,
     REQUEST_TYPE_EVAL,
     REQUEST_TYPE_AUTHENTICATE,
-    REQUEST_TYPE_JOIN,
-    REQUEST_TYPE_SUBSCRIBE,
     REQUEST_TYPE_ID,
     AUTH_TYPE_CHAP_SHA1,
     AUTH_TYPE_PAP_SHA256,
@@ -587,62 +581,6 @@ def __init__(self, conn, space_no, index_no, tuple_value, op_list):
         self._body = request_body
 
 
-class RequestJoin(Request):
-    """
-    Represents JOIN request.
-    """
-
-    request_type = REQUEST_TYPE_JOIN
-
-    def __init__(self, conn, server_uuid):
-        """
-        :param conn: Request sender.
-        :type conn: :class:`~tarantool.Connection`
-
-        :param server_uuid: UUID of connector "server".
-        :type server_uuid: :obj:`str`
-        """
-
-        super().__init__(conn)
-        request_body = self._dumps({IPROTO_SERVER_UUID: server_uuid})
-        self._body = request_body
-
-
-class RequestSubscribe(Request):
-    """
-    Represents SUBSCRIBE request.
-    """
-
-    request_type = REQUEST_TYPE_SUBSCRIBE
-
-    def __init__(self, conn, cluster_uuid, server_uuid, vclock):
-        """
-        :param conn: Request sender.
-        :type conn: :class:`~tarantool.Connection`
-
-        :param server_uuid: UUID of connector "server".
-        :type server_uuid: :obj:`str`
-
-        :param server_uuid: UUID of connector "server".
-        :type server_uuid: :obj:`str`
-
-        :param vclock: Connector "server" vclock.
-        :type vclock: :obj:`dict`
-
-        :raise: :exc:`~AssertionError`
-        """
-
-        super().__init__(conn)
-        assert isinstance(vclock, dict)
-
-        request_body = self._dumps({
-            IPROTO_CLUSTER_UUID: cluster_uuid,
-            IPROTO_SERVER_UUID: server_uuid,
-            IPROTO_VCLOCK: vclock
-        })
-        self._body = request_body
-
-
 class RequestOK(Request):
     """
     Represents OK acknowledgement.