From f2f2d07c28532587c7455884a86d80c89eec1c91 Mon Sep 17 00:00:00 2001
From: Gevorg Davoian <gdavoian@mirantis.com>
Date: Mon, 7 Mar 2016 17:09:31 +0200
Subject: [PATCH] Pika: fix sending fanout messages
This patch fixes the problem of sending fanout messages to listeners
located on the same host via the pika transport driver. Only one
listener per host received a fanout message before, because actually
there was only one queue per host, and that queue simply dispatched
incoming messages to listeners in a round robbin manner. In order to
fix this behaviour, every listener is supplied by its own unique
message queue bound to an appropriate fanout exchange now.
Since internal states of pika listeners (or pollers) were changed,
some tests related to pika pollers were also updated properly.
In addition, terms related to sending fanout messages to all_servers
listening some topic are refered to all_workers now (for better
description of what is really going on).
Closes-Bug: #1551661
Change-Id: I5aab6bee8758658fa3fa7e329b54ffa438e85f61
---
oslo_messaging/_drivers/impl_pika.py | 6 ++--
.../_drivers/pika_driver/pika_engine.py | 10 +++++--
.../_drivers/pika_driver/pika_poller.py | 16 +++++++---
.../tests/drivers/pika/test_poller.py | 29 +++++++++----------
4 files changed, 37 insertions(+), 24 deletions(-)
diff --git a/oslo_messaging/_drivers/impl_pika.py b/oslo_messaging/_drivers/impl_pika.py
index faa3e3bce..612c6054a 100644
--- a/oslo_messaging/_drivers/impl_pika.py
+++ b/oslo_messaging/_drivers/impl_pika.py
@@ -214,7 +214,7 @@ class PikaDriver(base.BaseDriver):
)
if target.fanout:
- return self.cast_all_servers(
+ return self.cast_all_workers(
exchange, target.topic, ctxt, message, expiration_time,
retrier
)
@@ -249,7 +249,7 @@ class PikaDriver(base.BaseDriver):
return reply.result
- def cast_all_servers(self, exchange, topic, ctxt, message, expiration_time,
+ def cast_all_workers(self, exchange, topic, ctxt, message, expiration_time,
retrier=None):
msg = pika_drv_msg.PikaOutgoingMessage(self._pika_engine, message,
ctxt)
@@ -257,7 +257,7 @@ class PikaDriver(base.BaseDriver):
msg.send(
exchange=exchange,
routing_key=self._pika_engine.get_rpc_queue_name(
- topic, "all_servers", retrier is None
+ topic, "all_workers", retrier is None
),
mandatory=False,
expiration_time=expiration_time,
diff --git a/oslo_messaging/_drivers/pika_driver/pika_engine.py b/oslo_messaging/_drivers/pika_driver/pika_engine.py
index a8c9a712f..bffd7ba8a 100644
--- a/oslo_messaging/_drivers/pika_driver/pika_engine.py
+++ b/oslo_messaging/_drivers/pika_driver/pika_engine.py
@@ -25,6 +25,8 @@ from pika import credentials as pika_credentials
import pika_pool
import six
+import uuid
+
from oslo_messaging._drivers.pika_driver import pika_exceptions as pika_drv_exc
LOG = logging.getLogger(__name__)
@@ -429,17 +431,21 @@ class PikaEngine(object):
return exchange or self.default_rpc_exchange
@staticmethod
- def get_rpc_queue_name(topic, server, no_ack):
+ def get_rpc_queue_name(topic, server, no_ack, worker=False):
"""Returns RabbitMQ queue name for given rpc request
:param topic: String, oslo.messaging target's topic
:param server: String, oslo.messaging target's server
:param no_ack: Boolean, use message delivery with acknowledges or not
+ :param worker: Boolean, use queue by single worker only or not
- :return: String, RabbitMQ exchange name
+ :return: String, RabbitMQ queue name
"""
queue_parts = ["no_ack" if no_ack else "with_ack", topic]
if server is not None:
queue_parts.append(server)
+ if worker:
+ queue_parts.append("worker")
+ queue_parts.append(uuid.uuid4().hex)
queue = '.'.join(queue_parts)
return queue
diff --git a/oslo_messaging/_drivers/pika_driver/pika_poller.py b/oslo_messaging/_drivers/pika_driver/pika_poller.py
index cc2d6e983..a5279fa8e 100644
--- a/oslo_messaging/_drivers/pika_driver/pika_poller.py
+++ b/oslo_messaging/_drivers/pika_driver/pika_poller.py
@@ -323,16 +323,24 @@ class RpcServicePikaPoller(PikaPoller):
queue=server_queue, routing_key=server_queue,
exchange_type='direct', queue_expiration=queue_expiration
)
- all_servers_routing_key = self._pika_engine.get_rpc_queue_name(
- self._target.topic, "all_servers", no_ack
+ queues_to_consume.append(
+ {"queue_name": server_queue, "no_ack": no_ack,
+ "consumer_tag": None}
+ )
+
+ worker_queue = self._pika_engine.get_rpc_queue_name(
+ self._target.topic, self._target.server, no_ack, True
+ )
+ all_workers_routing_key = self._pika_engine.get_rpc_queue_name(
+ self._target.topic, "all_workers", no_ack
)
self._pika_engine.declare_queue_binding_by_channel(
channel=self._channel, exchange=exchange, durable=False,
- queue=server_queue, routing_key=all_servers_routing_key,
+ queue=worker_queue, routing_key=all_workers_routing_key,
exchange_type='direct', queue_expiration=queue_expiration
)
queues_to_consume.append(
- {"queue_name": server_queue, "no_ack": no_ack,
+ {"queue_name": worker_queue, "no_ack": no_ack,
"consumer_tag": None}
)
diff --git a/oslo_messaging/tests/drivers/pika/test_poller.py b/oslo_messaging/tests/drivers/pika/test_poller.py
index 17ba3b708..1209d131a 100644
--- a/oslo_messaging/tests/drivers/pika/test_poller.py
+++ b/oslo_messaging/tests/drivers/pika/test_poller.py
@@ -228,9 +228,8 @@ class RpcServicePikaPollerTestCase(unittest.TestCase):
)
self._pika_engine.get_rpc_queue_name.side_effect = (
- lambda topic, server, no_ack: "_".join(
- [topic, str(server), str(no_ack)]
- )
+ lambda topic, server, no_ack, worker=False:
+ "_".join([topic, str(server), str(no_ack), str(worker)])
)
self._pika_engine.get_rpc_exchange_name.side_effect = (
@@ -277,49 +276,49 @@ class RpcServicePikaPollerTestCase(unittest.TestCase):
channel=self._poller_channel_mock, durable=False,
exchange="exchange",
exchange_type='direct',
- queue="topic_None_True",
+ queue="topic_None_True_False",
queue_expiration=12345,
- routing_key="topic_None_True"
+ routing_key="topic_None_True_False"
),
mock.call(
channel=self._poller_channel_mock, durable=False,
exchange="exchange",
exchange_type='direct',
- queue="topic_server_True",
+ queue="topic_server_True_False",
queue_expiration=12345,
- routing_key="topic_server_True"
+ routing_key="topic_server_True_False"
),
mock.call(
channel=self._poller_channel_mock, durable=False,
exchange="exchange",
exchange_type='direct',
- queue="topic_server_True",
+ queue="topic_server_True_True",
queue_expiration=12345,
- routing_key="topic_all_servers_True"
+ routing_key="topic_all_workers_True_False"
),
mock.call(
channel=self._poller_channel_mock, durable=False,
exchange="exchange",
exchange_type='direct',
- queue="topic_None_False",
+ queue="topic_None_False_False",
queue_expiration=12345,
- routing_key="topic_None_False"
+ routing_key="topic_None_False_False"
),
mock.call(
channel=self._poller_channel_mock, durable=False,
exchange="exchange",
exchange_type='direct',
- queue="topic_server_False",
+ queue="topic_server_False_False",
queue_expiration=12345,
- routing_key='topic_server_False'
+ routing_key='topic_server_False_False'
),
mock.call(
channel=self._poller_channel_mock, durable=False,
exchange="exchange",
exchange_type='direct',
- queue="topic_server_False",
+ queue="topic_server_False_True",
queue_expiration=12345,
- routing_key='topic_all_servers_False'
+ routing_key='topic_all_workers_False_False'
)
))