Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix RabbitMQ 'queue' argument usage #5492

Merged
merged 3 commits into from
Mar 25, 2020
Merged

fix RabbitMQ 'queue' argument usage #5492

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e83afe9
use default value of 'None' to avoid always picking default 'queues' …
wochinge Mar 25, 2020
8e60087
add changelog
wochinge Mar 25, 2020
985052e
Update rasa/core/brokers/pika.py
wochinge Mar 25, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions changelog/5492.bugfix.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
17 changes: 10 additions & 7 deletions rasa/core/brokers/pika.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
logger = logging.getLogger(__name__)

RABBITMQ_EXCHANGE = "rasa-exchange"
DEFAULT_QUEUE_NAME = "rasa_core_events"


def initialise_pika_connection(
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -326,12 +327,14 @@ 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` argument provided. It is suggested to explicitly "
f"specify a queue as described in {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"]
Expand Down Expand Up @@ -399,7 +402,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.

Expand Down
18 changes: 6 additions & 12 deletions tests/core/test_broker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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,
Expand All @@ -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")

Expand Down