From 37bd236a88f3fb01d61923bf9f701a441f80dead Mon Sep 17 00:00:00 2001 From: Nicholas Devenish Date: Fri, 23 Jan 2026 18:16:02 +0000 Subject: [PATCH] Make topic-sending recipes work This was apparently never used or tested? --- src/workflows/recipe/wrapper.py | 3 ++- src/workflows/transport/pika_transport.py | 4 +++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/src/workflows/recipe/wrapper.py b/src/workflows/recipe/wrapper.py index c15de95d..13af8c4b 100644 --- a/src/workflows/recipe/wrapper.py +++ b/src/workflows/recipe/wrapper.py @@ -265,9 +265,10 @@ def _send_to_destination( message = mangle_for_sending(message) self._retry_transport( broadcast, - self.recipe[destination]["topic"], + self.recipe[destination]["exchange"], message, headers=header, + topic=self.recipe[destination].get("topic"), **dest_kwargs, ) diff --git a/src/workflows/transport/pika_transport.py b/src/workflows/transport/pika_transport.py index 52c4cd42..aad4e823 100644 --- a/src/workflows/transport/pika_transport.py +++ b/src/workflows/transport/pika_transport.py @@ -543,6 +543,7 @@ def _broadcast( delay=None, expiration: int | None = None, transaction: int | None = None, + topic: str | None = None, **kwargs, ): """Send a message to a fanout exchange. @@ -554,6 +555,7 @@ def _broadcast( delay: Delay transport of message by this many seconds expiration: Optional TTL expiration time, in seconds, relative to sending time transaction: Transaction ID if message should be part of a transaction + topic: The routing key, if posting to a topic exchange kwargs: Arbitrary arguments for other transports. Ignored. """ assert delay is None, "Delay Not implemented" @@ -572,7 +574,7 @@ def _broadcast( self._pika_thread.send( exchange=destination, - routing_key="", + routing_key=topic or "", body=message, properties=properties, mandatory=False,