Add a _nowait variant to basic_publish

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,
         *,

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