Merge "Fix target resolution mismatch in neutron, nova, heat"

17 files changed, 286 insertions, 101 deletions

diff --git a/oslo_messaging/_drivers/impl_zmq.py b/oslo_messaging/_drivers/impl_zmq.py
index f91c341..aed6563 100644
--- a/oslo_messaging/_drivers/impl_zmq.py
+++ b/oslo_messaging/_drivers/impl_zmq.py
@@ -229,7 +229,7 @@ class ZmqDriver(base.BaseDriver):
         :param target: Message destination target
         :type target: oslo_messaging.Target
         """
-        server = self.server.get()
+        server = zmq_server.ZmqServer(self, self.conf, self.matchmaker)
         server.listen(target)
         return server
 
diff --git a/oslo_messaging/_drivers/zmq_driver/broker/zmq_broker.py b/oslo_messaging/_drivers/zmq_driver/broker/zmq_broker.py
index 5f20b80..8351e2e 100644
--- a/oslo_messaging/_drivers/zmq_driver/broker/zmq_broker.py
+++ b/oslo_messaging/_drivers/zmq_driver/broker/zmq_broker.py
@@ -16,7 +16,6 @@ import logging
 import os
 
 from oslo_utils import excutils
-import six
 from stevedore import driver
 
 from oslo_messaging._drivers.zmq_driver.broker import zmq_queue_proxy
@@ -51,11 +50,8 @@ class ZmqBroker(object):
         ).driver(self.conf)
 
         self.context = zmq.Context()
-        self.queue = six.moves.queue.Queue()
-        self.proxies = [zmq_queue_proxy.OutgoingQueueProxy(
-            conf, self.context, self.queue, self.matchmaker),
-            zmq_queue_proxy.IncomingQueueProxy(
-                conf, self.context, self.queue)
+        self.proxies = [zmq_queue_proxy.UniversalQueueProxy(
+            conf, self.context, self.matchmaker)
         ]
 
     def _create_ipc_dirs(self):
diff --git a/oslo_messaging/_drivers/zmq_driver/broker/zmq_queue_proxy.py b/oslo_messaging/_drivers/zmq_driver/broker/zmq_queue_proxy.py
index 11114d0..dd6665a 100644
--- a/oslo_messaging/_drivers/zmq_driver/broker/zmq_queue_proxy.py
+++ b/oslo_messaging/_drivers/zmq_driver/broker/zmq_queue_proxy.py
@@ -14,65 +14,69 @@
 
 import logging
 
-import six
-
 from oslo_messaging._drivers.zmq_driver.broker import zmq_base_proxy
 from oslo_messaging._drivers.zmq_driver.client.publishers\
     import zmq_dealer_publisher
 from oslo_messaging._drivers.zmq_driver import zmq_address
 from oslo_messaging._drivers.zmq_driver import zmq_async
+from oslo_messaging._drivers.zmq_driver import zmq_names
 from oslo_messaging._i18n import _LI
 
 zmq = zmq_async.import_zmq(zmq_concurrency='native')
 LOG = logging.getLogger(__name__)
 
 
-class OutgoingQueueProxy(zmq_base_proxy.BaseProxy):
+class UniversalQueueProxy(zmq_base_proxy.BaseProxy):
 
-    def __init__(self, conf, context, queue, matchmaker):
-        super(OutgoingQueueProxy, self).__init__(conf, context)
-        self.queue = queue
-        self.matchmaker = matchmaker
-        self.publisher = zmq_dealer_publisher.DealerPublisher(
-            conf, matchmaker)
-        LOG.info(_LI("Polling at outgoing proxy ..."))
+    def __init__(self, conf, context, matchmaker):
+        super(UniversalQueueProxy, self).__init__(conf, context)
+        self.poller = zmq_async.get_poller(zmq_concurrency='native')
 
-    def run(self):
-        try:
-            request = self.queue.get(timeout=self.conf.rpc_poll_timeout)
-            LOG.info(_LI("Redirecting request %s to TCP publisher ...")
-                     % request)
-            self.publisher.send_request(request)
-        except six.moves.queue.Empty:
-            return
+        self.router_socket = context.socket(zmq.ROUTER)
+        self.router_socket.bind(zmq_address.get_broker_address(conf))
 
+        self.poller.register(self.router_socket, self._receive_in_request)
+        LOG.info(_LI("Polling at universal proxy"))
 
-class IncomingQueueProxy(zmq_base_proxy.BaseProxy):
+        self.matchmaker = matchmaker
+        reply_receiver = zmq_dealer_publisher.ReplyReceiver(self.poller)
+        self.publisher = zmq_dealer_publisher.DealerPublisherProxy(
+            conf, matchmaker, reply_receiver)
 
-    def __init__(self, conf, context, queue):
-        super(IncomingQueueProxy, self).__init__(conf, context)
-        self.poller = zmq_async.get_poller(
-            zmq_concurrency='native')
+    def run(self):
+        message, socket = self.poller.poll(self.conf.rpc_poll_timeout)
+        if message is None:
+            return
 
-        self.queue = queue
+        if socket == self.router_socket:
+            self._redirect_in_request(message)
+        else:
+            self._redirect_reply(message)
 
-        self.socket = context.socket(zmq.ROUTER)
-        self.socket.bind(zmq_address.get_broker_address(conf))
-        self.poller.register(self.socket, self.receive_request)
-        LOG.info(_LI("Polling at incoming proxy ..."))
+    def _redirect_in_request(self, request):
+        LOG.info(_LI("-> Redirecting request %s to TCP publisher")
+                 % request)
+        self.publisher.send_request(request)
 
-    def run(self):
-        request, socket = self.poller.poll(self.conf.rpc_poll_timeout)
-        if request is None:
+    def _redirect_reply(self, reply):
+        LOG.info(_LI("Reply proxy %s") % reply)
+        if reply[zmq_names.IDX_REPLY_TYPE] == zmq_names.ACK_TYPE:
+            LOG.info(_LI("Acknowledge dropped %s") % reply)
             return
 
-        LOG.info(_LI("Received request and queue it: %s") % str(request))
+        LOG.info(_LI("<- Redirecting reply to ROUTER: reply: %s")
+                 % reply[zmq_names.IDX_REPLY_BODY:])
 
-        self.queue.put(request)
+        self.router_socket.send_multipart(reply[zmq_names.IDX_REPLY_BODY:])
 
-    def receive_request(self, socket):
+    def _receive_in_request(self, socket):
         reply_id = socket.recv()
         assert reply_id is not None, "Valid id expected"
         empty = socket.recv()
         assert empty == b'', "Empty delimiter expected"
-        return socket.recv_pyobj()
+        envelope = socket.recv_pyobj()
+        if envelope[zmq_names.FIELD_MSG_TYPE] == zmq_names.CALL_TYPE:
+            envelope[zmq_names.FIELD_REPLY_ID] = reply_id
+        payload = socket.recv_multipart()
+        payload.insert(0, envelope)
+        return payload
diff --git a/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_dealer_publisher.py b/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_dealer_publisher.py
index 2c8fc5e..602e5a9 100644
--- a/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_dealer_publisher.py
+++ b/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_dealer_publisher.py
@@ -29,12 +29,10 @@ class DealerPublisher(zmq_publisher_base.PublisherMultisend):
 
     def __init__(self, conf, matchmaker):
         super(DealerPublisher, self).__init__(conf, matchmaker, zmq.DEALER)
-        self.ack_receiver = AcknowledgementReceiver()
 
     def send_request(self, request):
 
-        if request.msg_type == zmq_names.CALL_TYPE:
-            raise zmq_publisher_base.UnsupportedSendPattern(request.msg_type)
+        self._check_request_pattern(request)
 
         dealer_socket, hosts = self._check_hosts_connections(request.target)
 
@@ -47,25 +45,26 @@ class DealerPublisher(zmq_publisher_base.PublisherMultisend):
                         % request.msg_type)
             return
 
-        self.ack_receiver.track_socket(dealer_socket.handle)
-
         if request.msg_type in zmq_names.MULTISEND_TYPES:
             for _ in range(dealer_socket.connections_count()):
                 self._send_request(dealer_socket, request)
         else:
             self._send_request(dealer_socket, request)
 
+    def _check_request_pattern(self, request):
+        if request.msg_type == zmq_names.CALL_TYPE:
+            raise zmq_publisher_base.UnsupportedSendPattern(request.msg_type)
+
     def _send_request(self, socket, request):
 
         socket.send(b'', zmq.SNDMORE)
         socket.send_pyobj(request)
 
-        LOG.info(_LI("Sending message %(message)s to a target %(target)s")
-                 % {"message": request.message,
+        LOG.info(_LI("Sending message_id %(message)s to a target %(target)s")
+                 % {"message": request.message_id,
                     "target": request.target})
 
     def cleanup(self):
-        self.ack_receiver.cleanup()
         super(DealerPublisher, self).cleanup()
 
 
@@ -81,7 +80,10 @@ class DealerPublisherLight(zmq_publisher_base.PublisherBase):
         if request.msg_type == zmq_names.CALL_TYPE:
             raise zmq_publisher_base.UnsupportedSendPattern(request.msg_type)
 
+        envelope = request.create_envelope()
+
         self.socket.send(b'', zmq.SNDMORE)
+        self.socket.send_pyobj(envelope, zmq.SNDMORE)
         self.socket.send_pyobj(request)
 
     def cleanup(self):
@@ -89,6 +91,67 @@ class DealerPublisherLight(zmq_publisher_base.PublisherBase):
         self.socket.close()
 
 
+class DealerPublisherProxy(DealerPublisher):
+
+    def __init__(self, conf, matchmaker, reply_receiver):
+        super(DealerPublisherProxy, self).__init__(conf, matchmaker)
+        self.reply_receiver = reply_receiver
+
+    def send_request(self, multipart_message):
+
+        envelope = multipart_message[zmq_names.MULTIPART_IDX_ENVELOPE]
+
+        LOG.info(_LI("Envelope: %s") % envelope)
+
+        target = envelope[zmq_names.FIELD_TARGET]
+        dealer_socket, hosts = self._check_hosts_connections(target)
+
+        if not dealer_socket.connections:
+            # NOTE(ozamiatin): Here we can provide
+            # a queue for keeping messages to send them later
+            # when some listener appears. However such approach
+            # being more reliable will consume additional memory.
+            LOG.warning(_LW("Request %s was dropped because no connection")
+                        % envelope[zmq_names.FIELD_MSG_TYPE])
+            return
+
+        self.reply_receiver.track_socket(dealer_socket.handle)
+
+        LOG.info(_LI("Sending message %(message)s to a target %(target)s")
+                 % {"message": envelope[zmq_names.FIELD_MSG_ID],
+                    "target": envelope[zmq_names.FIELD_TARGET]})
+
+        if envelope[zmq_names.FIELD_MSG_TYPE] in zmq_names.MULTISEND_TYPES:
+            for _ in range(dealer_socket.connections_count()):
+                self._send_request(dealer_socket, multipart_message)
+        else:
+            self._send_request(dealer_socket, multipart_message)
+
+    def _send_request(self, socket, multipart_message):
+
+        socket.send(b'', zmq.SNDMORE)
+        socket.send_pyobj(
+            multipart_message[zmq_names.MULTIPART_IDX_ENVELOPE],
+            zmq.SNDMORE)
+        socket.send(multipart_message[zmq_names.MULTIPART_IDX_BODY])
+
+
+class ReplyReceiver(object):
+
+    def __init__(self, poller):
+        self.poller = poller
+        LOG.info(_LI("Reply waiter created in broker"))
+
+    def _receive_reply(self, socket):
+        return socket.recv_multipart()
+
+    def track_socket(self, socket):
+        self.poller.register(socket, self._receive_reply)
+
+    def cleanup(self):
+        self.poller.close()
+
+
 class AcknowledgementReceiver(object):
 
     def __init__(self):
diff --git a/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_publisher_base.py b/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_publisher_base.py
index faee64d..0a8098a 100644
--- a/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_publisher_base.py
+++ b/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_publisher_base.py
@@ -89,6 +89,13 @@ class PublisherBase(object):
         :param request: Message data and destination container object
         :type request: zmq_request.Request
         """
+        LOG.info(_LI("Sending %(type)s message_id %(message)s to a target"
+                     "%(target)s key: %(key)s, host:%(host)s")
+                 % {"type": request.msg_type,
+                    "message": request.message_id,
+                    "target": request.target,
+                    "key": zmq_address.target_to_key(request.target),
+                    "host": request.host})
         socket.send_pyobj(request)
 
     def cleanup(self):
diff --git a/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_req_publisher.py b/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_req_publisher.py
index d4dbaa9..c6063b8 100644
--- a/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_req_publisher.py
+++ b/oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_req_publisher.py
@@ -14,6 +14,9 @@
 
 import contextlib
 import logging
+import uuid
+
+import six
 
 import oslo_messaging
 from oslo_messaging._drivers import common as rpc_common
@@ -40,24 +43,34 @@ class ReqPublisher(zmq_publisher_base.PublisherBase):
         if request.msg_type != zmq_names.CALL_TYPE:
             raise zmq_publisher_base.UnsupportedSendPattern(request.msg_type)
 
-        socket = self._connect_to_host(request.target, request.timeout)
+        socket, connect_address = self._connect_to_host(request.target,
+                                                        request.timeout)
+        request.host = connect_address
         self._send_request(socket, request)
         return self._receive_reply(socket, request)
 
+    def _resolve_host_address(self, target, timeout=0):
+        host = self.matchmaker.get_single_host(target, timeout)
+        return zmq_address.get_tcp_direct_address(host)
+
     def _connect_to_host(self, target, timeout=0):
 
         try:
             self.zmq_context = zmq.Context()
             socket = self.zmq_context.socket(zmq.REQ)
 
-            host = self.matchmaker.get_single_host(target, timeout)
-            connect_address = zmq_address.get_tcp_direct_address(host)
+            if six.PY3:
+                socket.setsockopt_string(zmq.IDENTITY, str(uuid.uuid1()))
+            else:
+                socket.identity = str(uuid.uuid1())
+
+            connect_address = self._resolve_host_address(target, timeout)
 
             LOG.info(_LI("Connecting REQ to %s") % connect_address)
 
             socket.connect(connect_address)
             self.outbound_sockets[str(target)] = socket
-            return socket
+            return socket, connect_address
 
         except zmq.ZMQError as e:
             errmsg = _LE("Error connecting to socket: %s") % str(e)
@@ -77,6 +90,7 @@ class ReqPublisher(zmq_publisher_base.PublisherBase):
             if reply is None:
                 raise oslo_messaging.MessagingTimeout(
                     "Timeout %s seconds was reached" % request.timeout)
+            LOG.info(_LI("Received reply %s") % reply)
             if reply[zmq_names.FIELD_FAILURE]:
                 raise rpc_common.deserialize_remote_exception(
                     reply[zmq_names.FIELD_FAILURE],
@@ -87,3 +101,26 @@ class ReqPublisher(zmq_publisher_base.PublisherBase):
     def close(self):
         # For contextlib compatibility
         self.cleanup()
+
+
+class ReqPublisherLight(ReqPublisher):
+
+    def __init__(self, conf, matchmaker):
+        super(ReqPublisherLight, self).__init__(conf, matchmaker)
+
+    def _resolve_host_address(self, target, timeout=0):
+        return zmq_address.get_broker_address(self.conf)
+
+    def _send_request(self, socket, request):
+
+        LOG.info(_LI("Sending %(type)s message_id %(message)s"
+                     " to a target %(target)s, host:%(host)s")
+                 % {"type": request.msg_type,
+                    "message": request.message_id,
+                    "target": request.target,
+                    "host": request.host})
+
+        envelope = request.create_envelope()
+
+        socket.send_pyobj(envelope, zmq.SNDMORE)
+        socket.send_pyobj(request)
diff --git a/oslo_messaging/_drivers/zmq_driver/client/zmq_client.py b/oslo_messaging/_drivers/zmq_driver/client/zmq_client.py
index 3e7888d..9fb3822 100644
--- a/oslo_messaging/_drivers/zmq_driver/client/zmq_client.py
+++ b/oslo_messaging/_drivers/zmq_driver/client/zmq_client.py
@@ -37,16 +37,18 @@ class ZmqClient(object):
         if self.conf.zmq_use_broker:
             self.dealer_publisher = zmq_dealer_publisher.DealerPublisherLight(
                 conf, zmq_address.get_broker_address(self.conf))
+            self.req_publisher_cls = zmq_req_publisher.ReqPublisherLight
         else:
             self.dealer_publisher = zmq_dealer_publisher.DealerPublisher(
                 conf, matchmaker)
+            self.req_publisher_cls = zmq_req_publisher.ReqPublisher
 
     def send_call(self, target, context, message, timeout=None, retry=None):
         with contextlib.closing(zmq_request.CallRequest(
                 target, context=context, message=message,
                 timeout=timeout, retry=retry,
                 allowed_remote_exmods=self.allowed_remote_exmods)) as request:
-            with contextlib.closing(zmq_req_publisher.ReqPublisher(
+            with contextlib.closing(self.req_publisher_cls(
                     self.conf, self.matchmaker)) as req_publisher:
                 return req_publisher.send_request(request)
 
diff --git a/oslo_messaging/_drivers/zmq_driver/client/zmq_request.py b/oslo_messaging/_drivers/zmq_driver/client/zmq_request.py
index 92d444a..455b7ba 100644
--- a/oslo_messaging/_drivers/zmq_driver/client/zmq_request.py
+++ b/oslo_messaging/_drivers/zmq_driver/client/zmq_request.py
@@ -63,6 +63,12 @@ class Request(object):
         self.message = message
         self.retry = retry
         self.message_id = str(uuid.uuid1())
+        self.proxy_reply_id = None
+
+    def create_envelope(self):
+        return {'msg_type': self.msg_type,
+                'message_id': self.message_id,
+                'target': self.target}
 
     @abc.abstractproperty
     def msg_type(self):
@@ -86,6 +92,11 @@ class RpcRequest(Request):
 
         super(RpcRequest, self).__init__(*args, **kwargs)
 
+    def create_envelope(self):
+        envelope = super(RpcRequest, self).create_envelope()
+        envelope['timeout'] = self.timeout
+        return envelope
+
 
 class CallRequest(RpcRequest):
 
diff --git a/oslo_messaging/_drivers/zmq_driver/matchmaker/matchmaker_redis.py b/oslo_messaging/_drivers/zmq_driver/matchmaker/matchmaker_redis.py
index c8402c6..cbf4e10 100644
--- a/oslo_messaging/_drivers/zmq_driver/matchmaker/matchmaker_redis.py
+++ b/oslo_messaging/_drivers/zmq_driver/matchmaker/matchmaker_redis.py
@@ -17,6 +17,7 @@ from oslo_config import cfg
 from oslo_utils import importutils
 
 from oslo_messaging._drivers.zmq_driver.matchmaker import base
+from oslo_messaging._drivers.zmq_driver import zmq_address
 
 redis = importutils.try_import('redis')
 LOG = logging.getLogger(__name__)
@@ -48,34 +49,30 @@ class RedisMatchMaker(base.MatchMakerBase):
             password=self.conf.matchmaker_redis.password,
         )
 
-    def _target_to_key(self, target):
-        attributes = ['topic', 'exchange', 'server']
-        prefix = "ZMQ-target"
-        key = ":".join((getattr(target, attr) or "*") for attr in attributes)
-        return "%s-%s" % (prefix, key)
-
-    def _get_keys_by_pattern(self, pattern):
-        return self._redis.keys(pattern)
-
     def _get_hosts_by_key(self, key):
         return self._redis.lrange(key, 0, -1)
 
     def register(self, target, hostname):
-        key = self._target_to_key(target)
-        if hostname not in self._get_hosts_by_key(key):
-            self._redis.lpush(key, hostname)
+
+        if target.topic and target.server:
+            key = zmq_address.target_to_key(target)
+            if hostname not in self._get_hosts_by_key(key):
+                self._redis.lpush(key, hostname)
+
+        if target.topic:
+            if hostname not in self._get_hosts_by_key(target.topic):
+                self._redis.lpush(target.topic, hostname)
+
+        if target.server:
+            if hostname not in self._get_hosts_by_key(target.server):
+                self._redis.lpush(target.server, hostname)
 
     def unregister(self, target, hostname):
-        key = self._target_to_key(target)
+        key = zmq_address.target_to_key(target)
         self._redis.lrem(key, 0, hostname)
 
     def get_hosts(self, target):
-        pattern = self._target_to_key(target)
-        if "*" not in pattern:
-            # pattern have no placeholders, so this is valid key
-            return self._get_hosts_by_key(pattern)
-
         hosts = []
-        for key in self._get_keys_by_pattern(pattern):
-            hosts.extend(self._get_hosts_by_key(key))
+        key = zmq_address.target_to_key(target)
+        hosts.extend(self._get_hosts_by_key(key))
         return hosts
diff --git a/oslo_messaging/_drivers/zmq_driver/poller/threading_poller.py b/oslo_messaging/_drivers/zmq_driver/poller/threading_poller.py
index c0a46d9..518d32e 100644
--- a/oslo_messaging/_drivers/zmq_driver/poller/threading_poller.py
+++ b/oslo_messaging/_drivers/zmq_driver/poller/threading_poller.py
@@ -38,12 +38,17 @@ class ThreadingPoller(zmq_poller.ZmqPoller):
         self.recv_methods = {}
 
     def register(self, socket, recv_method=None):
+        if socket in self.recv_methods:
+            return
         if recv_method is not None:
             self.recv_methods[socket] = recv_method
         self.poller.register(socket, zmq.POLLIN)
 
     def poll(self, timeout=None):
-        timeout *= 1000  # zmq poller waits milliseconds
+
+        if timeout:
+            timeout *= 1000  # zmq poller waits milliseconds
+
         sockets = None
 
         try:
diff --git a/oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_consumer_base.py b/oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_consumer_base.py
index 8bb2461..f002c45 100644
--- a/oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_consumer_base.py
+++ b/oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_consumer_base.py
@@ -19,6 +19,7 @@ import six
 
 from oslo_messaging._drivers import common as rpc_common
 from oslo_messaging._drivers.zmq_driver import zmq_async
+from oslo_messaging._drivers.zmq_driver import zmq_names
 from oslo_messaging._drivers.zmq_driver import zmq_socket
 from oslo_messaging._i18n import _LE, _LI
 
@@ -44,7 +45,7 @@ class ConsumerBase(object):
             self.sockets.append(socket)
             self.poller.register(socket, self.receive_message)
             LOG.info(_LI("Run %(stype)s consumer on %(addr)s:%(port)d"),
-                     {"stype": socket_type,
+                     {"stype": zmq_names.socket_type_str(socket_type),
                       "addr": socket.bind_address,
                       "port": socket.port})
             return socket
diff --git a/oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_router_consumer.py b/oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_router_consumer.py
index f601660..9c529da 100644
--- a/oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_router_consumer.py
+++ b/oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_router_consumer.py
@@ -43,11 +43,7 @@ class RouterIncomingMessage(base.IncomingMessage):
         """Reply is not needed for non-call messages"""
 
     def acknowledge(self):
-        LOG.info("Sending acknowledge for %s", self.msg_id)
-        ack_message = {zmq_names.FIELD_ID: self.msg_id}
-        self.socket.send(self.reply_id, zmq.SNDMORE)
-        self.socket.send(b'', zmq.SNDMORE)
-        self.socket.send_pyobj(ack_message)
+        LOG.info("Not sending acknowledge for %s", self.msg_id)
 
     def requeue(self):
         """Requeue is not supported"""
@@ -61,11 +57,11 @@ class RouterConsumer(zmq_consumer_base.SingleSocketConsumer):
         self.targets = []
         self.host = zmq_address.combine_address(self.conf.rpc_zmq_host,
                                                 self.port)
+        LOG.info("[%s] Run ROUTER consumer" % self.host)
 
     def listen(self, target):
 
-        LOG.info("Listen to target %s on %s:%d" %
-                 (target, self.address, self.port))
+        LOG.info("[%s] Listen to target %s" % (self.host, target))
 
         self.targets.append(target)
         self.matchmaker.register(target=target,
@@ -76,21 +72,25 @@ class RouterConsumer(zmq_consumer_base.SingleSocketConsumer):
         for target in self.targets:
             self.matchmaker.unregister(target, self.host)
 
+    def _receive_request(self, socket):
+        reply_id = socket.recv()
+        empty = socket.recv()
+        assert empty == b'', 'Bad format: empty delimiter expected'
+        request = socket.recv_pyobj()
+        return request, reply_id
+
     def receive_message(self, socket):
         try:
-            reply_id = socket.recv()
-            empty = socket.recv()
-            assert empty == b'', 'Bad format: empty delimiter expected'
-            request = socket.recv_pyobj()
-
-            LOG.info(_LI("Received %(msg_type)s message %(msg)s")
-                     % {"msg_type": request.msg_type,
-                        "msg": str(request.message)})
+            request, reply_id = self._receive_request(socket)
+            LOG.info(_LI("[%(host)s] Received %(type)s, %(id)s, %(target)s")
+                     % {"host": self.host,
+                        "type": request.msg_type,
+                        "id": request.message_id,
+                        "target": request.target})
 
             if request.msg_type == zmq_names.CALL_TYPE:
                 return zmq_incoming_message.ZmqIncomingRequest(
-                    self.server, request.context, request.message, socket,
-                    reply_id, self.poller)
+                    self.server, socket, reply_id, request, self.poller)
             elif request.msg_type in zmq_names.NON_BLOCKING_TYPES:
                 return RouterIncomingMessage(
                     self.server, request.context, request.message, socket,
@@ -100,3 +100,20 @@ class RouterConsumer(zmq_consumer_base.SingleSocketConsumer):
 
         except zmq.ZMQError as e:
             LOG.error(_LE("Receiving message failed: %s") % str(e))
+
+
+class RouterConsumerBroker(RouterConsumer):
+
+    def __init__(self, conf, poller, server):
+        super(RouterConsumerBroker, self).__init__(conf, poller, server)
+
+    def _receive_request(self, socket):
+        reply_id = socket.recv()
+        empty = socket.recv()
+        assert empty == b'', 'Bad format: empty delimiter expected'
+        envelope = socket.recv_pyobj()
+        request = socket.recv_pyobj()
+
+        if zmq_names.FIELD_REPLY_ID in envelope:
+            request.proxy_reply_id = envelope[zmq_names.FIELD_REPLY_ID]
+        return request, reply_id
diff --git a/oslo_messaging/_drivers/zmq_driver/server/zmq_incoming_message.py b/oslo_messaging/_drivers/zmq_driver/server/zmq_incoming_message.py
index f43ec23..5e93249 100644
--- a/oslo_messaging/_drivers/zmq_driver/server/zmq_incoming_message.py
+++ b/oslo_messaging/_drivers/zmq_driver/server/zmq_incoming_message.py
@@ -28,10 +28,12 @@ zmq = zmq_async.import_zmq()
 
 class ZmqIncomingRequest(base.IncomingMessage):
 
-    def __init__(self, listener, context, message, socket, rep_id, poller):
-        super(ZmqIncomingRequest, self).__init__(listener, context, message)
+    def __init__(self, listener, socket, rep_id, request, poller):
+        super(ZmqIncomingRequest, self).__init__(listener, request.context,
+                                                 request.message)
         self.reply_socket = socket
         self.reply_id = rep_id
+        self.request = request
         self.received = None
         self.poller = poller
 
@@ -39,15 +41,21 @@ class ZmqIncomingRequest(base.IncomingMessage):
         if failure is not None:
             failure = rpc_common.serialize_remote_exception(failure,
                                                             log_failure)
-        message_reply = {zmq_names.FIELD_REPLY: reply,
+        message_reply = {zmq_names.FIELD_TYPE: zmq_names.REPLY_TYPE,
+                         zmq_names.FIELD_REPLY: reply,
                          zmq_names.FIELD_FAILURE: failure,
-                         zmq_names.FIELD_LOG_FAILURE: log_failure}
+                         zmq_names.FIELD_LOG_FAILURE: log_failure,
+                         zmq_names.FIELD_ID: self.request.proxy_reply_id}
 
-        LOG.info("Replying %s REP", (str(message_reply)))
+        LOG.info("Replying %s REP", (str(self.request.message_id)))
 
         self.received = True
         self.reply_socket.send(self.reply_id, zmq.SNDMORE)
         self.reply_socket.send(b'', zmq.SNDMORE)
+        if self.request.proxy_reply_id:
+            self.reply_socket.send_string(zmq_names.REPLY_TYPE, zmq.SNDMORE)
+            self.reply_socket.send(self.request.proxy_reply_id, zmq.SNDMORE)
+            self.reply_socket.send(b'', zmq.SNDMORE)
         self.reply_socket.send_pyobj(message_reply)
         self.poller.resume_polling(self.reply_socket)
 
diff --git a/oslo_messaging/_drivers/zmq_driver/server/zmq_server.py b/oslo_messaging/_drivers/zmq_driver/server/zmq_server.py
index afe03b8..0680e7f 100644
--- a/oslo_messaging/_drivers/zmq_driver/server/zmq_server.py
+++ b/oslo_messaging/_drivers/zmq_driver/server/zmq_server.py
@@ -31,8 +31,12 @@ class ZmqServer(base.Listener):
         super(ZmqServer, self).__init__(driver)
         self.matchmaker = matchmaker
         self.poller = zmq_async.get_poller()
-        self.rpc_consumer = zmq_router_consumer.RouterConsumer(
-            conf, self.poller, self)
+        if conf.zmq_use_broker:
+            self.rpc_consumer = zmq_router_consumer.RouterConsumerBroker(
+                conf, self.poller, self)
+        else:
+            self.rpc_consumer = zmq_router_consumer.RouterConsumer(
+                conf, self.poller, self)
         self.notify_consumer = self.rpc_consumer
         self.consumers = [self.rpc_consumer]
 
diff --git a/oslo_messaging/_drivers/zmq_driver/zmq_address.py b/oslo_messaging/_drivers/zmq_driver/zmq_address.py
index e8c4829..afc9249 100644
--- a/oslo_messaging/_drivers/zmq_driver/zmq_address.py
+++ b/oslo_messaging/_drivers/zmq_driver/zmq_address.py
@@ -27,3 +27,14 @@ def get_tcp_random_address(conf):
 
 def get_broker_address(conf):
     return "ipc://%s/zmq-broker" % conf.rpc_zmq_ipc_dir
+
+
+def target_to_key(target):
+    if target.topic and target.server:
+        attributes = ['topic', 'server']
+        key = ".".join(getattr(target, attr) for attr in attributes)
+        return key
+    if target.topic:
+        return target.topic
+    if target.server:
+        return target.server
diff --git a/oslo_messaging/_drivers/zmq_driver/zmq_names.py b/oslo_messaging/_drivers/zmq_driver/zmq_names.py
index a317456..f7401ab 100644
--- a/oslo_messaging/_drivers/zmq_driver/zmq_names.py
+++ b/oslo_messaging/_drivers/zmq_driver/zmq_names.py
@@ -17,10 +17,23 @@ from oslo_messaging._drivers.zmq_driver import zmq_async
 zmq = zmq_async.import_zmq()
 
 
+FIELD_TYPE = 'type'
 FIELD_FAILURE = 'failure'
 FIELD_REPLY = 'reply'
 FIELD_LOG_FAILURE = 'log_failure'
 FIELD_ID = 'id'
+FIELD_MSG_ID = 'message_id'
+FIELD_MSG_TYPE = 'msg_type'
+FIELD_REPLY_ID = 'reply_id'
+FIELD_TARGET = 'target'
+
+
+IDX_REPLY_TYPE = 1
+IDX_REPLY_BODY = 2
+
+MULTIPART_IDX_ENVELOPE = 0
+MULTIPART_IDX_BODY = 1
+
 
 CALL_TYPE = 'call'
 CAST_TYPE = 'cast'
@@ -28,6 +41,9 @@ CAST_FANOUT_TYPE = 'cast-f'
 NOTIFY_TYPE = 'notify'
 NOTIFY_FANOUT_TYPE = 'notify-f'
 
+REPLY_TYPE = 'reply'
+ACK_TYPE = 'ack'
+
 MESSAGE_TYPES = (CALL_TYPE,
                  CAST_TYPE,
                  CAST_FANOUT_TYPE,
diff --git a/oslo_messaging/_drivers/zmq_driver/zmq_socket.py b/oslo_messaging/_drivers/zmq_driver/zmq_socket.py
index 2a4144c..8e51e30 100644
--- a/oslo_messaging/_drivers/zmq_driver/zmq_socket.py
+++ b/oslo_messaging/_drivers/zmq_driver/zmq_socket.py
@@ -57,6 +57,9 @@ class ZmqSocket(object):
     def send_pyobj(self, *args, **kwargs):
         self.handle.send_pyobj(*args, **kwargs)
 
+    def send_multipart(self, *args, **kwargs):
+        self.handle.send_multipart(*args, **kwargs)
+
     def recv(self, *args, **kwargs):
         return self.handle.recv(*args, **kwargs)
 
@@ -69,6 +72,9 @@ class ZmqSocket(object):
     def recv_pyobj(self, *args, **kwargs):
         return self.handle.recv_pyobj(*args, **kwargs)
 
+    def recv_multipart(self, *args, **kwargs):
+        return self.handle.recv_multipart(*args, **kwargs)
+
     def close(self, *args, **kwargs):
         self.handle.close(*args, **kwargs)