Merge pull request RasaHQ#5492 from RasaHQ/pika-old-queue-arg-fix

fix RabbitMQ 'queue' argument usage

Author: wochinge

changelog/5492.bugfix.rst

@@ -0,0 +1,4 @@
+Fix an issue where the deprecated ``queue`` parameter for the :ref:`event-brokers-pika`
+was ignored and Rasa Open Source published the events to the ``rasa_core_events``
+queue instead. Note that this does not change the fact that the ``queue`` argument
+is deprecated in favor of the ``queues`` argument.

rasa/core/brokers/pika.py

@@ -27,6 +27,7 @@
 logger = logging.getLogger(__name__)
 
 RABBITMQ_EXCHANGE = "rasa-exchange"
+DEFAULT_QUEUE_NAME = "rasa_core_events"
 
 
 def initialise_pika_connection(
@@ -224,7 +225,7 @@ def __init__(
         username: Text,
         password: Text,
         port: Union[int, Text] = 5672,
-        queues: Union[List[Text], Tuple[Text], Text, None] = ("rasa_core_events",),
+        queues: Union[List[Text], Tuple[Text], Text, None] = None,
         should_keep_unpublished_messages: bool = True,
         raise_on_failure: bool = False,
         log_level: Union[Text, int] = os.environ.get(
@@ -278,7 +279,7 @@ def rasa_environment(self) -> Optional[Text]:
 
     @staticmethod
     def _get_queues_from_args(
-        queues_arg: Union[List[Text], Tuple[Text], Text, None], kwargs: Any,
+        queues_arg: Union[List[Text], Tuple[Text], Text, None], kwargs: Any
     ) -> Union[List[Text], Tuple[Text]]:
         """Get queues for this event broker.
 
@@ -326,12 +327,15 @@ def _get_queues_from_args(
         if queue_arg:
             return queue_arg  # pytype: disable=bad-return-type
 
-        raise ValueError(
-            f"Could not initialise `PikaEventBroker` due to invalid "
-            f"`queues` or `queue` argument in constructor. See "
-            f"{DOCS_URL_PIKA_EVENT_BROKER}."
+        raise_warning(
+            f"No `queues` or `queue` argument provided. It is suggested to "
+            f"explicitly specify a queue as described in "
+            f"{DOCS_URL_PIKA_EVENT_BROKER}. "
+            f"Using the default queue '{DEFAULT_QUEUE_NAME}' for now."
         )
 
+        return [DEFAULT_QUEUE_NAME]
+
     @classmethod
     def from_endpoint_config(
         cls, broker_config: Optional["EndpointConfig"]
@@ -399,7 +403,7 @@ def _run_pika_io_loop(self) -> None:
         self._pika_connection.ioloop.start()
 
     def is_ready(
-        self, attempts: int = 1000, wait_time_between_attempts_in_seconds: float = 0.01,
+        self, attempts: int = 1000, wait_time_between_attempts_in_seconds: float = 0.01
     ) -> bool:
         """Spin until the pika channel is open.
 

tests/core/test_broker.py

@@ -8,7 +8,7 @@
 from rasa.core.brokers.broker import EventBroker
 from rasa.core.brokers.file import FileEventBroker
 from rasa.core.brokers.kafka import KafkaEventBroker
-from rasa.core.brokers.pika import PikaEventBroker
+from rasa.core.brokers.pika import PikaEventBroker, DEFAULT_QUEUE_NAME
 from rasa.core.brokers.sql import SQLEventBroker
 from rasa.core.events import Event, Restarted, SlotSet, UserUttered
 from rasa.utils.endpoints import EndpointConfig, read_endpoint_config
@@ -53,9 +53,9 @@ def test_pika_message_property_app_id(monkeypatch: MonkeyPatch):
     "queue_arg,queues_arg,expected,warning",
     [
         # default case
-        (None, ["rasa_core_events"], ["rasa_core_events"], None),
+        (None, ["q1"], ["q1"], None),
         # only provide `queue`
-        ("rasa_core_events", None, ["rasa_core_events"], FutureWarning),
+        ("q1", None, ["q1"], FutureWarning),
         # supplying a list for `queue` works too
         (["q1", "q2"], None, ["q1", "q2"], FutureWarning),
         # `queues` arg supplied, takes precedence
@@ -64,11 +64,13 @@ def test_pika_message_property_app_id(monkeypatch: MonkeyPatch):
         ("q1", ["q2", "q3"], ["q2", "q3"], FutureWarning),
         # only supplying `queues` works, and queues is a string
         (None, "q1", ["q1"], None),
+        # no queues provided. Use default queue and print warning.
+        (None, None, [DEFAULT_QUEUE_NAME], UserWarning),
     ],
 )
 def test_pika_queues_from_args(
-    queues_arg: Union[Text, List[Text], None],
     queue_arg: Union[Text, List[Text], None],
+    queues_arg: Union[Text, List[Text], None],
     expected: List[Text],
     warning: Optional[Type[Warning]],
     monkeypatch: MonkeyPatch,
@@ -82,14 +84,6 @@ def test_pika_queues_from_args(
     assert pika_producer.queues == expected
 
 
-def test_pika_invalid_queues_argument(monkeypatch: MonkeyPatch):
-    # patch PikaEventBroker so it doesn't try to connect to RabbitMQ on init
-    monkeypatch.setattr(PikaEventBroker, "_run_pika", lambda _: None)
-
-    with pytest.raises(ValueError):
-        PikaEventBroker("", "", "", queues=None, queue=None)
-
-
 def test_no_broker_in_config():
     cfg = read_endpoint_config(DEFAULT_ENDPOINTS_FILE, "event_broker")