diff --git a/src/workflows/transport/pika_transport.py b/src/workflows/transport/pika_transport.py index 4cbd706..7c86915 100644 --- a/src/workflows/transport/pika_transport.py +++ b/src/workflows/transport/pika_transport.py @@ -801,8 +801,12 @@ def __init__( self._channel_has_active_tx: dict[BlockingChannel, bool] = {} # Information on whether a channel is transactional self._channel_is_transactional: dict[BlockingChannel, bool] = {} - # A common, shared channel, used for sending messages outside of transactions. - self._pika_shared_channel: BlockingChannel | None + # A common, shared channel, used for transactions not tied to a subscription. + self._pika_shared_channel: BlockingChannel | None = None + # A channel in publisher-confirms mode, used for sending messages outside + # of transactions. Confirm and transaction modes are mutually exclusive + # on a channel, so this has to be kept separate from the shared channel. + self._pika_confirm_channel: BlockingChannel | None = None # Are we allowed to reconnect. Can only be turned off, never on self._reconnection_allowed: bool = True # Our list of connection parameters, so we know where to connect to @@ -1117,7 +1121,9 @@ def _send() -> None: channel = self._transaction_on_channel.inverse[transaction_id] self._channel_has_active_tx[channel] = True else: - channel = self._get_shared_channel() + channel = self._get_confirm_channel() + # On a confirm-mode channel this blocks until the broker + # has acknowledged the message channel.basic_publish( exchange=exchange, routing_key=routing_key, @@ -1126,6 +1132,14 @@ def _send() -> None: mandatory=mandatory, ) future.set_result(None) + except ( + pika.exceptions.UnroutableError, + pika.exceptions.NackError, + ) as e: + # The broker rejected this message. This is a problem for + # the sender, not for the connection, so don't propagate + # it into (and terminate) the connection runloop. + future.set_exception(e) except BaseException as e: future.set_exception(e) raise @@ -1367,9 +1381,18 @@ def _get_shared_channel(self) -> BlockingChannel: if not self._pika_shared_channel: self._pika_shared_channel = self._connection.channel() - ##### self._pika_shared_channel.confirm_delivery() return self._pika_shared_channel + def _get_confirm_channel(self) -> BlockingChannel: + """Get the publisher-confirms channel. Create if necessary.""" + + assert self._connection is not None + + if not self._pika_confirm_channel: + self._pika_confirm_channel = self._connection.channel() + self._pika_confirm_channel.confirm_delivery() + return self._pika_confirm_channel + def _recreate_subscriptions(self) -> None: """Resubscribe all existing subscriptions""" old_subscriptions = self._subscriptions @@ -1481,6 +1504,7 @@ def _run(self) -> None: # Clear the channels because this might be a reconnect self._pika_channels = bidict() self._pika_shared_channel = None + self._pika_confirm_channel = None self._transaction_on_channel = bidict() self._channel_has_active_tx = {} self._channel_is_transactional = {} diff --git a/tests/transport/test_pika.py b/tests/transport/test_pika.py index 516c21e..ea998cf 100644 --- a/tests/transport/test_pika.py +++ b/tests/transport/test_pika.py @@ -1068,16 +1068,70 @@ def test_pikathread_send(connection_params, test_channel): ).result() # But should fail with it declared - pytest.xfail("UnroutableError is not raised without publisher confirms, #96") with pytest.raises(pika.exceptions.UnroutableError): thread.send( "", "unroutable-missing-queue", "Another Message", mandatory=True ).result() + # A rejected message must not take down the connection + assert thread.connection_alive + thread.send("", queue, "After unroutable").result() + assert test_channel.basic_get(queue, auto_ack=True)[2] == b"After unroutable" + finally: thread.join(stop=True) +def test_pikathread_send_confirmed_during_shared_transaction( + connection_params, test_channel +): + """Confirmed sends must coexist with a transaction on the shared channel.""" + queue = test_channel.temporary_queue_declare(exclusive=True) + thread = _PikaThread(connection_params) + try: + thread.start() + thread.wait_for_connection() + + # Open a transaction on the shared (non-subscription) channel + thread.tx_select(transaction_id=1, subscription_id=None).result() + thread.send("", queue, "in transaction", transaction_id=1).result() + + # A non-transactional send is confirmed and delivered immediately + thread.send("", queue, "outside transaction").result() + assert test_channel.basic_get(queue, auto_ack=True)[2] == b"outside transaction" + # ...while the transactional message is still held back + assert test_channel.basic_get(queue, auto_ack=True)[0] is None + + thread.tx_commit(transaction_id=1).result() + assert test_channel.basic_get(queue, auto_ack=True)[2] == b"in transaction" + + # Unroutable messages are still caught while the transaction is open + thread.tx_select(transaction_id=2, subscription_id=None).result() + with pytest.raises(pika.exceptions.UnroutableError): + thread.send( + "", "unroutable-missing-queue", "Another Message", mandatory=True + ).result() + thread.tx_rollback(transaction_id=2).result() + assert thread.connection_alive + finally: + thread.join(re_raise=True, stop=True) + + +def test_pikatransport_send_unroutable_raises(connection_params): + """PikaTransport.send uses mandatory=True, so a missing queue now raises.""" + transport = PikaTransport() + with mock.patch.object( + transport, "_generate_connection_parameters", return_value=connection_params + ): + transport.connect() + try: + with pytest.raises(pika.exceptions.UnroutableError): + transport.send("unroutable-missing-queue", "message") + assert transport.is_connected() + finally: + transport.disconnect() + + def test_pikathread_send_named_exchange(connection_params, test_channel): exchange = test_channel.temporary_exchange_declare( exchange_type="topic", auto_delete=True