Add support for NATS JetStream as a transport

docs/reference/kombu.transport.nats.rst

@@ -0,0 +1,31 @@
+================================================
+ NATS Transport - ``kombu.transport.nats``
+================================================
+
+.. currentmodule:: kombu.transport.nats
+
+.. automodule:: kombu.transport.nats
+
+    .. contents::
+        :local:
+
+    Transport
+    ---------
+
+    .. autoclass:: Transport
+        :members:
+        :undoc-members:
+
+    Channel
+    -------
+
+    .. autoclass:: Channel
+        :members:
+        :undoc-members:
+
+    Message
+    -------
+
+    .. autoclass:: Message
+        :members:
+        :undoc-members:

examples/nats_receive.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import sys
+from pprint import pformat
+
+from kombu import Connection, Consumer, Exchange, Queue, eventloop
+
+LOCAL_SERVER = "localhost"
+DEMO_SERVER = "demo.nats.io"
+
+server = LOCAL_SERVER
+use_demo_server = len(sys.argv) > 1 and sys.argv[1] == "--demo"
+if use_demo_server:
+    server = DEMO_SERVER
+
+
+exchange = Exchange("exchange", "direct", durable=False)
+msg_queue = Queue("kombu_demo", 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(f"nats://{server}: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,30 @@
+from __future__ import annotations
+
+import sys
+
+from nats.js.api import StorageType
+
+from kombu import Connection, Exchange, Queue
+
+LOCAL_SERVER = "localhost"
+DEMO_SERVER = "demo.nats.io"
+
+server = LOCAL_SERVER
+use_demo_server = len(sys.argv) > 1 and sys.argv[1] == "--demo"
+if use_demo_server:
+    server = DEMO_SERVER
+
+
+exchange = Exchange("exchange", "direct", durable=False)
+msg_queue = Queue("kombu_demo", exchange=exchange, routing_key="messages")
+
+
+with Connection(f"nats://{server}:4222", transport_options={
+    "stream_config": {
+        "storage": StorageType.FILE,
+    }
+}) 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:Transport',
 }
 
 _transport_cache = {}