From 49f9429911b13dc43bc93ea6fd4f5f38dfddb8ee Mon Sep 17 00:00:00 2001
From: Mehdi Abaakouk <mehdi.abaakouk@enovance.com>
Date: Tue, 22 Jul 2014 09:42:52 -0700
Subject: [PATCH 5/5] Declare DirectPublisher exchanges with passive=True

If rabbit dies, the consumer can be disconnected before the publisher
sends, and if the consumer hasn't delared the queue, the publisher's
will send a message to an exchange that's not bound to a queue, and
the message wll be lost.  Setting passive=True will cause the
publisher to fail and retry if the consumer hasn't declared the
receiving queue yet.

Co-Authored-By: Noel Burton-Krahn <noel@burton-krahn.com>
Closes-Bug: #1338732
Change-Id: I5ba4d311b97236d3a85a9f5badff61f12b08c12d
---
 oslo/messaging/_drivers/impl_rabbit.py | 55 ++++++++++++++++++++++++++++++++--
 tests/drivers/test_impl_rabbit.py      | 23 ++++++++++++--
 2 files changed, 74 insertions(+), 4 deletions(-)

diff --git a/oslo/messaging/_drivers/impl_rabbit.py b/oslo/messaging/_drivers/impl_rabbit.py
index 61e8fd1..629753e 100644
--- a/oslo/messaging/_drivers/impl_rabbit.py
+++ b/oslo/messaging/_drivers/impl_rabbit.py
@@ -110,6 +110,31 @@ rabbit_opts = [
 LOG = logging.getLogger(__name__)
 
 
+class _DecayingTimer(object):
+    def __init__(self, duration=None):
+        self._duration = duration
+        self._ends_at = None
+
+    def start(self):
+        if self._duration is not None:
+            self._ends_at = time.time() + max(0, self._duration)
+
+    def check_return(self, timeout_callback=None, *args, **kwargs):
+        maximum = kwargs.pop('maximum', None)
+
+        if self._duration is None:
+            return None if maximum is None else maximum
+        if self._ends_at is None:
+            raise RuntimeError(_("Can not check/return a timeout from a timer"
+                               " that has not been started."))
+
+        left = self._ends_at - time.time()
+        if left <= 0 and timeout_callback is not None:
+            timeout_callback(*args, **kwargs)
+
+        return left if maximum is None else min(left, maximum)
+
+
 def _get_queue_arguments(conf):
     """Construct the arguments for declaring a queue.
 
@@ -356,7 +381,8 @@ class DirectPublisher(Publisher):
 
         options = {'durable': False,
                    'auto_delete': True,
-                   'exclusive': False}
+                   'exclusive': False,
+                   'passive': True}
         options.update(kwargs)
         super(DirectPublisher, self).__init__(channel, topic, topic,
                                               type='direct', **options)
@@ -372,6 +398,7 @@ class TopicPublisher(Publisher):
         options = {'durable': conf.amqp_durable_queues,
                    'auto_delete': conf.amqp_auto_delete,
                    'exclusive': False}
+
         options.update(kwargs)
         super(TopicPublisher, self).__init__(channel,
                                              exchange_name,
@@ -739,7 +766,31 @@ class Connection(object):
 
     def direct_send(self, msg_id, msg):
         """Send a 'direct' message."""
-        self.publisher_send(DirectPublisher, msg_id, msg)
+
+        timer = _DecayingTimer(duration=60)
+        timer.start()
+        # NOTE(sileht): retry at least 60sec, after we have a good change
+        # that the caller is really dead too...
+
+        while True:
+            try:
+                self.publisher_send(DirectPublisher, msg_id, msg)
+            except self.connection.channel_errors as exc:
+                # NOTE(noelbk/sileht):
+                # If rabbit dies, the consumer can be disconnected before the
+                # publisher sends, and if the consumer hasn't delared the
+                # queue, the publisher's will send a message to an exchange
+                # that's not bound to a queue, and the message wll be lost.
+                # So we set passive=True to the publisher exchange and catch
+                # the 404 kombu ChannelError and retry until the exchange
+                # appears
+                if exc.code == 404 and timer.check_return() > 0:
+                    LOG.info(_("The exchange to reply to %s doesn't "
+                               "exist yet, retrying...") % msg_id)
+                    time.sleep(1)
+                    continue
+                raise
+            return
 
     def topic_send(self, exchange_name, topic, msg, timeout=None, retry=None):
         """Send a 'topic' message."""
diff --git a/tests/drivers/test_impl_rabbit.py b/tests/drivers/test_impl_rabbit.py
index 3a3f340..29dfb1e 100644
--- a/tests/drivers/test_impl_rabbit.py
+++ b/tests/drivers/test_impl_rabbit.py
@@ -216,8 +216,27 @@ class TestSendReceive(test_utils.BaseTestCase):
                         raise ZeroDivisionError
                     except Exception:
                         failure = sys.exc_info()
-                    msgs[i].reply(failure=failure,
-                                  log_failure=not self.expected)
+
+                    # NOTE(noelbk) confirm that Publisher exchanges
+                    # are always declared with passive=True
+                    outer_self = self
+                    test_exchange_was_called = [False]
+                    old_init = kombu.entity.Exchange.__init__
+
+                    def new_init(self, *args, **kwargs):
+                        test_exchange_was_called[0] = True
+                        outer_self.assertTrue(kwargs['passive'])
+                        old_init(self, *args, **kwargs)
+                    kombu.entity.Exchange.__init__ = new_init
+
+                    try:
+                        msgs[i].reply(failure=failure,
+                                      log_failure=not self.expected)
+                    finally:
+                        kombu.entity.Exchange.__init__ = old_init
+
+                    self.assertTrue(test_exchange_was_called[0])
+
                 elif self.rx_id:
                     msgs[i].reply({'rx_id': i})
                 else:
-- 
2.3.0

