diff --git a/aiormq/channel.py b/aiormq/channel.py index 9d7d576..8541a6d 100644 --- a/aiormq/channel.py +++ b/aiormq/channel.py @@ -698,6 +698,50 @@ async def basic_publish( return await countdown(confirmation) + def basic_publish_nowait( + self, + body: bytes, + *, + exchange: str = "", + routing_key: str = "", + properties: Optional[spec.Basic.Properties] = None, + mandatory: bool = False, + immediate: bool = False, + ) -> None: + assert not self.publisher_confirms, "need await for publisher_confirm" + + _check_routing_key(routing_key) + + publish_frame = spec.Basic.Publish( + exchange=exchange, + routing_key=routing_key, + mandatory=mandatory, + immediate=immediate, + ) + + content_header = ContentHeader( + properties=properties or spec.Basic.Properties(delivery_mode=1), + body_size=len(body), + ) + + if not content_header.properties.message_id: + # UUID compatible random bytes + rnd_uuid = UUID(int=getrandbits(128), version=4) + content_header.properties.message_id = rnd_uuid.hex + + self.delivery_tag += 1 + + body_frames: List[Union[FrameType, ContentBody]] + body_frames = [publish_frame, content_header] + body_frames += self._split_body(body) + + self.write_queue.put_nowait( + ChannelFrame.marshall( + frames=body_frames, + channel_number=self.number, + ), + ) + async def basic_qos( self, *, diff --git a/tests/test_channel.py b/tests/test_channel.py index b9652e7..4a927b0 100644 --- a/tests/test_channel.py +++ b/tests/test_channel.py @@ -38,6 +38,37 @@ async def test_simple(amqp_channel: aiormq.Channel): assert message.body == b"foo bar" +async def test_simple_nowait(amqp_channel: aiormq.Channel): + if amqp_channel.publisher_confirms: + pytest.skip("nowait makes no sense with publisher_confirms") + await amqp_channel.basic_qos(prefetch_count=1) + assert amqp_channel.number + + queue = asyncio.Queue() + + deaclare_ok = await amqp_channel.queue_declare(auto_delete=True) + consume_ok = await amqp_channel.basic_consume(deaclare_ok.queue, queue.put) + amqp_channel.basic_publish_nowait( + b"foo", + routing_key=deaclare_ok.queue, + properties=aiormq.spec.Basic.Properties(message_id="123"), + ) + + message: DeliveredMessage = await queue.get() + assert message.body == b"foo" + + cancel_ok = await amqp_channel.basic_cancel(consume_ok.consumer_tag) + assert cancel_ok.consumer_tag == consume_ok.consumer_tag + assert cancel_ok.consumer_tag not in amqp_channel.consumers + await amqp_channel.queue_delete(deaclare_ok.queue) + + deaclare_ok = await amqp_channel.queue_declare(auto_delete=True) + amqp_channel.basic_publish_nowait(b"foo bar", routing_key=deaclare_ok.queue) + + message = await amqp_channel.basic_get(deaclare_ok.queue, no_ack=True) + assert message.body == b"foo bar" + + async def test_blank_body(amqp_channel: aiormq.Channel): await amqp_channel.basic_qos(prefetch_count=1) assert amqp_channel.number