[zmq] Refactoring of zmq client

Split ZmqClient in order to create different clients
for different configurations (direct/proxy).

Change-Id: Ib374f62f53f2c82278ce5bc555ea440e8eba6617

oslo_messaging/_drivers/impl_zmq.py

@@ -182,12 +182,18 @@ class ZmqDriver(base.BaseDriver):
             self.get_matchmaker_backend(url),
         ).driver(self.conf, url=url)
 
+        client_cls = zmq_client.ZmqClientProxy
+        if conf.use_pub_sub and not conf.use_router_proxy:
+            client_cls = zmq_client.ZmqClientMixDirectPubSub
+        elif not conf.use_pub_sub and not conf.use_router_proxy:
+            client_cls = zmq_client.ZmqClientDirect
+
         self.client = LazyDriverItem(
-            zmq_client.ZmqClient, self.conf, self.matchmaker,
+            client_cls, self.conf, self.matchmaker,
             self.allowed_remote_exmods)
 
         self.notifier = LazyDriverItem(
-            zmq_client.ZmqClient, self.conf, self.matchmaker,
+            client_cls, self.conf, self.matchmaker,
             self.allowed_remote_exmods)
 
         super(ZmqDriver, self).__init__(conf, url, default_exchange,

oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_publisher_base.py

@@ -112,30 +112,37 @@ class SocketsManager(object):
         self.socket_to_publishers = None
         self.socket_to_routers = None
 
-    def _track_socket(self, socket, target):
-        self.outbound_sockets[str(target)] = (socket, time.time())
-
     def get_hosts(self, target):
         return self.matchmaker.get_hosts(
             target, zmq_names.socket_type_str(self.listener_type))
 
+    @staticmethod
+    def _key_from_target(target):
+        return target.topic if target.fanout else str(target)
+
     def _get_hosts_and_connect(self, socket, target):
         hosts = self.get_hosts(target)
         self._connect_to_hosts(socket, target, hosts)
 
+    def _track_socket(self, socket, target):
+        key = self._key_from_target(target)
+        self.outbound_sockets[key] = (socket, time.time())
+
     def _connect_to_hosts(self, socket, target, hosts):
         for host in hosts:
             socket.connect_to_host(host)
         self._track_socket(socket, target)
 
     def _check_for_new_hosts(self, target):
-        socket, tm = self.outbound_sockets[str(target)]
+        key = self._key_from_target(target)
+        socket, tm = self.outbound_sockets[key]
         if 0 <= self.conf.zmq_target_expire <= time.time() - tm:
             self._get_hosts_and_connect(socket, target)
         return socket
 
     def get_socket(self, target):
-        if str(target) in self.outbound_sockets:
+        key = self._key_from_target(target)
+        if key in self.outbound_sockets:
             socket = self._check_for_new_hosts(target)
         else:
             socket = zmq_socket.ZmqSocket(self.conf, self.zmq_context,
@@ -143,16 +150,6 @@ class SocketsManager(object):
             self._get_hosts_and_connect(socket, target)
         return socket
 
-    def get_socket_to_hosts(self, target, hosts):
-        key = str(target)
-        if key in self.outbound_sockets:
-            socket, tm = self.outbound_sockets[key]
-        else:
-            socket = zmq_socket.ZmqSocket(self.conf, self.zmq_context,
-                                          self.socket_type)
-            self._connect_to_hosts(socket, target, hosts)
-        return socket
-
     def get_socket_to_publishers(self):
         if self.socket_to_publishers is not None:
             return self.socket_to_publishers

oslo_messaging/_drivers/zmq_driver/client/zmq_client.py

@@ -13,6 +13,7 @@
 #    under the License.
 
 
+from oslo_messaging._drivers import common
 from oslo_messaging._drivers.zmq_driver.client.publishers.dealer \
     import zmq_dealer_call_publisher
 from oslo_messaging._drivers.zmq_driver.client.publishers.dealer \
@@ -28,45 +29,102 @@ from oslo_messaging._drivers.zmq_driver import zmq_names
 zmq = zmq_async.import_zmq()
 
 
-class ZmqClient(zmq_client_base.ZmqClientBase):
+class WrongClientException(common.RPCException):
+    """Raised if client type doesn't match configuration"""
+
+
+class ZmqClientMixDirectPubSub(zmq_client_base.ZmqClientBase):
+    """Client for using with direct connections and fanout over proxy:
+
+        use_pub_sub = true
+        use_router_proxy = false
+
+    """
 
     def __init__(self, conf, matchmaker=None, allowed_remote_exmods=None):
 
+        if conf.use_router_proxy or not conf.use_pub_sub:
+            raise WrongClientException()
+
         self.sockets_manager = zmq_publisher_base.SocketsManager(
             conf, matchmaker, zmq.ROUTER, zmq.DEALER)
 
-        default_publisher = zmq_dealer_publisher.DealerPublisher(
-            conf, matchmaker)
-
-        publisher_to_proxy = zmq_dealer_publisher_proxy.DealerPublisherProxy(
+        fanout_publisher = zmq_dealer_publisher_proxy.DealerPublisherProxy(
             conf, matchmaker, self.sockets_manager.get_socket_to_publishers())
 
-        call_publisher = zmq_dealer_publisher_proxy.DealerCallPublisherProxy(
-            conf, matchmaker, self.sockets_manager) if conf.use_router_proxy \
-            else zmq_dealer_call_publisher.DealerCallPublisher(
-                conf, matchmaker, self.sockets_manager)
+        super(ZmqClientMixDirectPubSub, self).__init__(
+            conf, matchmaker, allowed_remote_exmods,
+            publishers={
+                zmq_names.CALL_TYPE:
+                    zmq_dealer_call_publisher.DealerCallPublisher(
+                        conf, matchmaker, self.sockets_manager),
+
+                zmq_names.CAST_FANOUT_TYPE: fanout_publisher,
+
+                zmq_names.NOTIFY_TYPE: fanout_publisher,
+
+                "default": zmq_dealer_publisher.DealerPublisherAsync(
+                    conf, matchmaker)
+            }
+        )
+
+
+class ZmqClientDirect(zmq_client_base.ZmqClientBase):
+    """This kind of client (publishers combination) is to be used for
+    direct connections only:
+
+        use_pub_sub = false
+        use_router_proxy = false
+    """
 
-        cast_publisher = publisher_to_proxy if conf.use_router_proxy \
-            else zmq_dealer_publisher.DealerPublisherAsync(
-                conf, matchmaker)
+    def __init__(self, conf, matchmaker=None, allowed_remote_exmods=None):
 
-        fanout_publisher = publisher_to_proxy \
-            if conf.use_pub_sub else default_publisher
+        if conf.use_pub_sub or conf.use_router_proxy:
+            raise WrongClientException()
 
-        super(ZmqClient, self).__init__(
+        self.sockets_manager = zmq_publisher_base.SocketsManager(
+            conf, matchmaker, zmq.ROUTER, zmq.DEALER)
+
+        super(ZmqClientDirect, self).__init__(
             conf, matchmaker, allowed_remote_exmods,
             publishers={
-                zmq_names.CALL_TYPE: call_publisher,
+                zmq_names.CALL_TYPE:
+                    zmq_dealer_call_publisher.DealerCallPublisher(
+                        conf, matchmaker, self.sockets_manager),
 
-                zmq_names.CAST_TYPE: cast_publisher,
+                "default": zmq_dealer_publisher.DealerPublisher(
+                    conf, matchmaker)
+            }
+        )
 
-                # Here use DealerPublisherLight for sending request to proxy
-                # which finally uses PubPublisher to send fanout in case of
-                # 'use_pub_sub' option configured.
-                zmq_names.CAST_FANOUT_TYPE: fanout_publisher,
 
-                zmq_names.NOTIFY_TYPE: fanout_publisher,
+class ZmqClientProxy(zmq_client_base.ZmqClientBase):
+    """Client for using with proxy:
+
+        use_pub_sub = true
+        use_router_proxy = true
+    or
+        use_pub_sub = false
+        use_router_proxy = true
+    """
+
+    def __init__(self, conf, matchmaker=None, allowed_remote_exmods=None):
+
+        if not conf.use_router_proxy:
+            raise WrongClientException()
+
+        self.sockets_manager = zmq_publisher_base.SocketsManager(
+            conf, matchmaker, zmq.ROUTER, zmq.DEALER)
+
+        super(ZmqClientProxy, self).__init__(
+            conf, matchmaker, allowed_remote_exmods,
+            publishers={
+                zmq_names.CALL_TYPE:
+                    zmq_dealer_publisher_proxy.DealerCallPublisherProxy(
+                        conf, matchmaker, self.sockets_manager),
 
-                "default": default_publisher
+                "default": zmq_dealer_publisher_proxy.DealerPublisherProxy(
+                        conf, matchmaker,
+                        self.sockets_manager.get_socket_to_publishers())
             }
         )

oslo_messaging/_drivers/zmq_driver/matchmaker/matchmaker_redis.py

@@ -190,7 +190,7 @@ class RedisMatchMaker(base.MatchMakerBase):
         key = zmq_address.target_to_key(target, listener_type)
         hosts.extend(self._get_hosts_by_key(key))
 
-        if not hosts and target.topic and target.server:
+        if (not hosts or target.fanout) and target.topic and target.server:
             key = zmq_address.prefix_str(target.topic, listener_type)
             hosts.extend(self._get_hosts_by_key(key))
 

setup-test-env-zmq.sh

@@ -16,6 +16,7 @@ cat > ${DATADIR}/zmq.conf <<EOF
 transport_url=${TRANSPORT_URL}
 rpc_zmq_matchmaker=${ZMQ_MATCHMAKER}
 rpc_zmq_ipc_dir=${ZMQ_IPC_DIR}
+use_pub_sub=true
 use_router_proxy=true
 [matchmaker_redis]
 port=${ZMQ_REDIS_PORT}