Repository: nats-io/nats.py (language: python)

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
```
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 / hints:
Likely suspect:
[nats/js/client.py](https://github.com/nats-io/nats.py/blob/455bdf3759032ca5ade80029bc9a930a36c97a51/nats/js/client.py#L402C1-L405C1)

Likely suspect:
[nats/js/client.py](https://github.com/nats-io/nats.py/blob/455bdf3759032ca5ade80029bc9a930a36c97a51/nats/js/client.py#L402C1-L405C1)


Fix patch:
diff --git a/nats/js/client.py b/nats/js/client.py
index bc951be1..fac3fc73 100644
--- a/nats/js/client.py
+++ b/nats/js/client.py
@@ -414,9 +414,10 @@ async def cb(msg):
             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


Test patch:
diff --git a/tests/test_js.py b/tests/test_js.py
index 720c9819..9c14bb0d 100644
--- a/tests/test_js.py
+++ b/tests/test_js.py
@@ -1807,6 +1807,50 @@ async def cb(msg):
         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):