Merge branch 'subscribe-bind'

nats/js/client.py

@@ -128,13 +128,16 @@ async def subscribe(
         durable: Optional[str] = None,
         stream: Optional[str] = None,
         config: Optional[api.ConsumerConfig] = None,
-        manual_ack: Optional[bool] = False,
-        ordered_consumer: Optional[bool] = False,
+        manual_ack: bool = False,
+        ordered_consumer: bool = False,
         idle_heartbeat: Optional[float] = None,
-        flow_control: Optional[bool] = False,
+        flow_control: bool = False,
     ) -> Subscription:
-        """
-        subscribe returns a `Subscription` that is bound to a push based consumer.
+        """Create consumer if needed and push-subscribe to it.
+
+        1. Check if consumer exists.
+        2. Creates consumer if needed.
+        3. Calls `subscribe_bind`.
 
         :param subject: Subject from a stream from JetStream.
         :param queue: Deliver group name from a set a of queue subscribers.
@@ -271,17 +274,36 @@ async def cb(msg):
             consumer_info = await self._jsm.add_consumer(stream, config=config)
             consumer = consumer_info.name
 
+        if consumer is None:
+            raise TypeError("cannot detect consumer")
+        if config is None:
+            raise TypeError("config is required for existing durable consumer")
+        return await self.subscribe_bind(
+            cb=cb,
+            stream=stream,
+            config=config,
+            manual_ack=manual_ack,
+            ordered_consumer=ordered_consumer,
+            consumer=consumer,
+        )
+
+    async def subscribe_bind(
+        self,
+        stream: str,
+        config: api.ConsumerConfig,
+        consumer: str,
+        cb: Optional[Callback] = None,
+        manual_ack: bool = False,
+        ordered_consumer: bool = False,
+    ) -> Subscription:
+        """Push-subscribe to an existing consumer.
+        """
         # By default, async subscribers wrap the original callback and
         # auto ack the messages as they are delivered.
         if cb and not manual_ack:
             cb = self._auto_ack_callback(cb)
-
-        # TODO (@orsinium): too many assumptions, refactor the code above to ensure they hold true.
-        assert config is not None
         if config.deliver_subject is None:
-            raise TypeError("config.deliver_subject is required")
-        if consumer is None:
-            raise TypeError("cannot detect consumer")
+            config.deliver_subject = self._nc.new_inbox()
         sub = await self._nc.subscribe(
             subject=config.deliver_subject,
             queue=config.deliver_group or "",

tests/test_js.py

@@ -866,6 +866,50 @@ async def test_ephemeral_subscribe(self):
         assert len(info2.name) > 0
         assert info1.name != info2.name
 
+    @async_test
+    async def test_subscribe_bind(self):
+        nc = await nats.connect()
+        js = nc.jetstream()
+
+        stream_name = "hello-stream"
+        subject_name = "hello-subject"
+        consumer_name = "alice"
+        await js.add_stream(name=stream_name, subjects=[subject_name])
+
+        # Create the consumer and assign a deliver subject which
+        # will then be picked up on bind.
+        inbox = nc.new_inbox()
+        config = nats.js.api.ConsumerConfig(deliver_subject=inbox)
+        consumer_info = await js.add_consumer(
+            stream=stream_name,
+            config=config,
+            durable_name=consumer_name,
+        )
+        assert consumer_info.stream_name == stream_name
+        assert consumer_info.name == consumer_name
+        assert consumer_info.config.durable_name == consumer_name
+
+        # Subscribe using the deliver subject that was chosen before.
+        sub = await js.subscribe_bind(
+            stream=consumer_info.stream_name,
+            consumer=consumer_info.name,
+            config=consumer_info.config,
+        )
+        for i in range(10):
+            await js.publish(subject_name, f'Hello World {i}'.encode())
+
+        msgs = []
+        for i in range(0, 10):
+            msg = await sub.next_msg()
+            msgs.append(msg)
+            await msg.ack()
+        assert len(msgs) == 10
+        assert sub.pending_msgs == 0
+
+        info = await sub.consumer_info()
+        assert info.num_ack_pending == 0
+        assert info.num_pending == 0
+
 
 class AckPolicyTest(SingleJetStreamServerTestCase):