Add NATS demo server to examples

Author: joeriddles

examples/nats_receive.py

@@ -1,11 +1,21 @@
 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("queue", exchange=exchange, routing_key="messages")
+msg_queue = Queue("kombu_demo", exchange=exchange, routing_key="messages")
 
 
 def pretty(obj):
@@ -19,7 +29,7 @@ def process_msg(body, message):
     message.ack()
 
 
-with Connection("nats://localhost:4222") as connection:
+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

@@ -1,12 +1,29 @@
 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("queue", exchange=exchange, routing_key="messages")
+msg_queue = Queue("kombu_demo", exchange=exchange, routing_key="messages")
 
 
-with Connection("nats://localhost:4222") as conn:
+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/nats_jetstream.py

@@ -191,7 +191,6 @@ def _ensure_stream(self, queue):
 
         stream_config = StreamConfig(
             name=stream_name,
-            # subjects=[f"{queue}.>"],
             subjects=[queue],
             retention=RetentionPolicy.WORK_QUEUE,
             max_consumers=-1,
@@ -227,7 +226,6 @@ def _ensure_consumer(self, queue):
             durable_name=consumer_name,
             deliver_policy=DeliverPolicy.ALL,
             ack_policy=AckPolicy.EXPLICIT,
-            # filter_subject=f"{queue}.>",
             filter_subject=queue,
         )
 
@@ -242,7 +240,6 @@ def _ensure_consumer(self, queue):
     def _put(self, queue, message, **kwargs):
         """Put a message on a queue."""
         self._ensure_stream(queue)
-        # subject = f"{queue}.{message.get('id', '')}"
         subject = queue
         self._event_loop.run_until_complete(
             self._js.publish(subject, str_to_bytes(dumps(message)))
@@ -256,7 +253,6 @@ def _get(self, queue, **kwargs):
         try:
             pull_sub = self._event_loop.run_until_complete(
                 self._js.pull_subscribe(
-                    # f"{queue}.>",
                     queue,
                     self._get_consumer_name(queue),
                     stream=self._get_stream_name(queue),