Repository: nats-io/nats.py

Issue:
JetstreamContext.subscribe() ignores deliver_subject from config argument
### Observed behavior

When `.subscribe()` of `JetstreamContext` is called, and the specified consumer does not exist, the client creates a consumer based on the given config object. However, attribute deliver_subject of config object is ignored, and random inbox subject is used instead.
Creating consumer works fine when using JetstreamManager.

### Expected behavior

Created consumer should use specified deliver_subject.

### Server and client version

Server version 2.10.21
Client version 2.9.0

### Host environment

_No response_

### Steps to reproduce

1. Create a stream named "test".
2. Run the following code:
```python
async def main():
    nc = await connect("nats://localhost:4222")
    js = nc.jetstream()

    deliver_subject = "stream-test.consume3"
    cc = ConsumerConfig(
        name="consumer3",
        deliver_subject=deliver_subject,
        ack_policy=AckPolicy.ALL,
        deliver_policy=DeliverPolicy.NEW,
        durable_name="consumer3",
    )

    sub = await js.subscribe("", stream="test", durable="consumer3", cb=cb, config=cc)
    print("Subscribed")
    while True:
        await asyncio.sleep(5)

```
3. Verify created consumer `nats consumer info test consumer3`:
```
Information for Consumer test > consumer3 created 2024-09-29T15:45:46+02:00

Configuration:

                Name: consumer3
    Delivery Subject: _INBOX.rvFDfLPnsefqQD5R3OOsOT
      Deliver Policy: New
          Ack Policy: All
            Ack Wait: 30s
       Replay Policy: Instant
     Max Ack Pending: 1,000
        Flow Control: false

State:

   Last Delivered Message: Consumer sequence: 0 Stream sequence: 113
     Acknowledgment floor: Consumer sequence: 0 Stream sequence: 0
         Outstanding Acks: 0 out of maximum 1,000
     Redelivered Messages: 0
     Unprocessed Messages: 0
          Active Interest: Active
```

Discussion:

Patch:
diff --git a/nats/js/client.py b/nats/js/client.py
index bc951be..7958dbb 100644
--- a/nats/js/client.py
+++ b/nats/js/client.py
@@ -414,9 +414,10 @@ class JetStreamContext(JetStreamManager):
             if inactive_threshold:
                 config.inactive_threshold = inactive_threshold
 
-            # Create inbox for push consumer.
-            deliver = self._nc.new_inbox()
-            config.deliver_subject = deliver
+            # Create inbox for push consumer, if deliver_subject is not assigned already.
+            if config.deliver_subject is None:
+                deliver = self._nc.new_inbox()
+                config.deliver_subject = deliver
 
             # Auto created consumers use the filter subject.
             config.filter_subject = subject
@@ -1244,7 +1245,7 @@ class JetStreamContext(JetStreamManager):
                         )
                         msg = await self._sub.next_msg(timeout=deadline)
                         status = JetStreamContext.is_status_msg(msg)
-                        if status == api.StatusCode.NO_MESSAGES:
+                        if status == api.StatusCode.NO_MESSAGES or status == api.StatusCode.REQUEST_TIMEOUT:
                             # No more messages after this so fallthrough
                             # after receiving the rest.
                             break


Test patch:
diff --git a/tests/test_js.py b/tests/test_js.py
index 720c981..9c14bb0 100644
--- a/tests/test_js.py
+++ b/tests/test_js.py
@@ -1807,6 +1807,50 @@ class SubscribeTest(SingleJetStreamServerTestCase):
         self.assertEqual(coroutines_before, coroutines_after_unsubscribe)
         self.assertNotEqual(coroutines_before, coroutines_after_subscribe)
 
+    @async_test
+    async def test_subscribe_push_config(self):
+        nc = await nats.connect()
+        js = nc.jetstream()
+
+        await js.add_stream(name="pconfig", subjects=["pconfig"])
+
+        s, d = ([], [])
+
+        async def cb_s(msg):
+            s.append(msg.data)
+
+        async def cb_d(msg):
+            d.append(msg.data)
+
+        #Create config for our subscriber
+        cc = nats.js.api.ConsumerConfig(
+            name="pconfig-ps", deliver_subject="pconfig-deliver"
+        )
+
+        #Make stream consumer with set deliver_subjct
+        sub_s = await js.subscribe(
+            "pconfig", stream="pconfig", cb=cb_s, config=cc
+        )
+        #Make direct sub on deliver_subject
+        sub_d = await nc.subscribe("pconfig-deliver", "check-queue", cb=cb_d)
+
+        #Stream consumer sub should have configured subject
+        assert sub_s.subject == "pconfig-deliver"
+
+        #Publish some messages
+        for i in range(10):
+            await js.publish("pconfig", f'Hello World {i}'.encode())
+
+        await asyncio.sleep(0.5)
+        #Both subs should recieve same messages, but we are not sure about order
+        assert len(s) == len(d)
+        assert set(s) == set(d)
+
+        #Cleanup
+        await js.delete_consumer("pconfig", "pconfig-ps")
+        await js.delete_stream("pconfig")
+        await nc.close()
+
 
 class AckPolicyTest(SingleJetStreamServerTestCase):