Fix IPC direct topic routing.

Direct messages were being stripped of the host
value when performing IPC forwarding. This caused
direct topics to be round-robined to all
services running on the system consuming from
the same base topic name.

i.e. if 'scheduler.host1' and 'scheduler.host2'
were running on the SAME machine, messages to
'scheduler.host1' may have been routed to
'scheduler.host2'.

Now, mulitple processing specifying different
rpc_zmq_host parameters will consume on
separate direct topics and will not
round-robin to other processes.

Adds a zmq-specific test to ensure that messages to
directed topics are not consumed by other
consumers of direct topics sharing a bare
topic on the same host.

Fixes bug 1123715

Change-Id: I939c24397e58492fc16561666aed3ca891325e9c

2 files changed, 15 insertions, 3 deletions

diff --git a/tests/unit/rpc/common.py b/tests/unit/rpc/common.py
index 6359142..5a2af19 100644
--- a/tests/unit/rpc/common.py
+++ b/tests/unit/rpc/common.py
@@ -140,7 +140,8 @@ class BaseRpcTestCase(test_utils.BaseTestCase):
                                             "args": {"value": value}})
         self.assertEqual(self.context.to_dict(), result)
 
-    def _test_cast(self, method, value, args=None, fanout=False):
+    def _test_cast(self, method, value, args=None, fanout=False,
+                   topic_nested=None):
         """Test casts by pushing items through a channeled queue.
 
            @param: method a reference to a method returning a value
@@ -148,6 +149,8 @@ class BaseRpcTestCase(test_utils.BaseTestCase):
            @param: args optional dictionary arguments to method
            @param: fanout boolean for use of rpc fanout method
         """
+        topic_nested = topic_nested or self.topic_nested
+
         # Not a true global, but capitalized so
         # it is clear it is leaking scope into Nested()
         QUEUE = eventlet.queue.Queue()
@@ -163,7 +166,7 @@ class BaseRpcTestCase(test_utils.BaseTestCase):
                 QUEUE.put(method(*args, **kwargs))
 
         nested = Nested()
-        conn = self._create_consumer(nested, self.topic_nested, fanout)
+        conn = self._create_consumer(nested, topic_nested, fanout)
 
         rpc_method = (self.rpc.cast, self.rpc.fanout_cast)[fanout]
 
@@ -173,7 +176,7 @@ class BaseRpcTestCase(test_utils.BaseTestCase):
             msg['args'].update(args)
 
         rpc_method(FLAGS, self.context,
-                   self.topic_nested,
+                   topic_nested,
                    msg)
 
         try:
diff --git a/tests/unit/rpc/test_zmq.py b/tests/unit/rpc/test_zmq.py
index 7a1e501..c197c35 100644
--- a/tests/unit/rpc/test_zmq.py
+++ b/tests/unit/rpc/test_zmq.py
@@ -127,3 +127,12 @@ class RpcZmqDirectTopicTestCase(_RpcZmqBaseTestCase):
         super(RpcZmqDirectTopicTestCase, self).setUp(
             topic='test.127.0.0.1',
             topic_nested='nested.127.0.0.1')
+
+    def test_cast_wrong_direct_topic_failure(self):
+        try:
+            self._test_cast(common.TestReceiver.echo, 42, {"value": 42},
+                            fanout=False, topic_nested='nested.localhost')
+        except Exception:
+            return
+        self.expectFailure("Message should not have been consumed.",
+                           self.assertTrue, True)