js: add consumer config inactive threshold

Signed-off-by: Waldemar Quevedo <[email protected]>

Author: wallyqs

nats/js/api.py

@@ -394,6 +394,9 @@ class ConsumerConfig(Base):
     deliver_subject: Optional[str] = None
     deliver_group: Optional[str] = None
 
+    # Ephemeral inactivity threshold
+    inactive_threshold: Optional[float] = None  # in seconds
+
     # Generally inherited by parent stream and other markers, now can
     # be configured directly.
     num_replicas: Optional[int] = None
@@ -405,12 +408,16 @@ class ConsumerConfig(Base):
     def from_response(cls, resp: Dict[str, Any]):
         cls._convert_nanoseconds(resp, 'ack_wait')
         cls._convert_nanoseconds(resp, 'idle_heartbeat')
+        cls._convert_nanoseconds(resp, 'inactive_threshold')
         return super().from_response(resp)
 
     def as_dict(self) -> Dict[str, object]:
         result = super().as_dict()
         result['ack_wait'] = self._to_nanoseconds(self.ack_wait)
         result['idle_heartbeat'] = self._to_nanoseconds(self.idle_heartbeat)
+        result['inactive_threshold'] = self._to_nanoseconds(
+            self.inactive_threshold
+        )
         return result
 
 

tests/test_client.py

@@ -1840,7 +1840,6 @@ async def reconnected_cb():
             reconnected.set_result(True)
 
         async def err_cb(e):
-            print("ERROR: ", e)
             nonlocal errors
             errors.append(e)
 

tests/test_js.py

@@ -1091,6 +1091,26 @@ async def test_consumer_with_name(self):
         assert err.value.err_code == 10017
         assert err.value.description == 'consumer name in subject does not match durable name in request'
 
+        # Create ephemeral pull consumer with a name and inactive threshold.
+        stream_name = "ctests"
+        consumer_name = "inactive"
+        cinfo = await jsm.add_consumer(
+            stream_name,
+            name=consumer_name,
+            ack_policy="explicit",
+            inactive_threshold=2,  # seconds
+        )
+        assert cinfo.config.name == consumer_name
+
+        sub = await js.pull_subscribe_bind(consumer_name, stream_name)
+        msgs = await sub.fetch(1)
+        assert msgs[0].data == b'hello world!'
+        ok = await msgs[0].ack_sync()
+        assert ok
+
+        cinfo = await sub.consumer_info()
+        assert cinfo.config.inactive_threshold == 2.0
+
         await nc.close()