Implement basic NATS JetStream transport

Author: joeriddles

examples/nats_receive.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from pprint import pformat
+
+from kombu import Connection, Consumer, Exchange, Queue, eventloop
+
+exchange = Exchange("exchange", "direct", durable=False)
+msg_queue = Queue("queue", exchange=exchange, routing_key="messages")
+
+
+def pretty(obj):
+    return pformat(obj, indent=4)
+
+
+def process_msg(body, message):
+    print(f"Received message: {body!r}")
+    print(f"  properties:\n{pretty(message.properties)}")
+    print(f"  delivery_info:\n{pretty(message.delivery_info)}")
+    message.ack()
+
+
+with Connection("nats://localhost:4222") as connection:
+    with Consumer(connection, msg_queue, callbacks=[process_msg]) as consumer:
+        for msg in eventloop(connection):
+            pass

examples/nats_send.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from kombu import Connection, Exchange, Queue
+
+exchange = Exchange("exchange", "direct", durable=False)
+msg_queue = Queue("queue", exchange=exchange, routing_key="messages")
+
+
+with Connection("nats://localhost:4222") as conn:
+    producer = conn.Producer()
+    producer.publish(
+        "hello world", exchange=exchange, routing_key="messages", declare=[msg_queue]
+    )

kombu/transport/__init__.py

@@ -45,6 +45,7 @@ def supports_librabbitmq() -> bool | None:
     'azureservicebus': 'kombu.transport.azureservicebus:Transport',
     'pyro': 'kombu.transport.pyro:Transport',
     'gcpubsub': 'kombu.transport.gcpubsub:Transport',
+    'nats': 'kombu.transport.nats_jetstream:Transport',
 }
 
 _transport_cache = {}