Backport sourcing-consumer config for ADR-60

nats/src/nats/js/api.py

@@ -23,6 +23,9 @@
 
 
 class Header(str, Enum):
+    BATCH_COMMIT = "Nats-Batch-Commit"
+    BATCH_ID = "Nats-Batch-Id"
+    BATCH_SEQUENCE = "Nats-Batch-Sequence"
     CONSUMER_STALLED = "Nats-Consumer-Stalled"
     DESCRIPTION = "Description"
     EXPECTED_LAST_MSG_ID = "Nats-Expected-Last-Msg-Id"
@@ -163,6 +166,25 @@ class PubAck(Base):
     seq: int
     domain: Optional[str] = None
     duplicate: Optional[bool] = None
+    batch_id: Optional[str] = None
+    batch_size: Optional[int] = None
+
+    @classmethod
+    def from_response(cls, resp: Dict[str, Any]) -> PubAck:
+        # Server uses ``batch``/``count`` for atomic batch publish (ADR-50).
+        if "batch" in resp and "batch_id" not in resp:
+            resp["batch_id"] = resp.pop("batch")
+        if "count" in resp and "batch_size" not in resp:
+            resp["batch_size"] = resp.pop("count")
+        return super().from_response(resp)
+
+    def as_dict(self) -> Dict[str, object]:
+        result = super().as_dict()
+        if "batch_id" in result:
+            result["batch"] = result.pop("batch_id")
+        if "batch_size" in result:
+            result["count"] = result.pop("batch_size")
+        return result
 
 
 @dataclass
@@ -183,6 +205,20 @@ def as_dict(self) -> Dict[str, object]:
         return result
 
 
+@dataclass
+class StreamConsumerSource(Base):
+    """Pre-created push-durable consumer used for stream sourcing/mirroring (ADR-60).
+
+    Required when sourcing or mirroring from a workqueue or interest stream so
+    the server can drive acknowledgements via flow control rather than
+    auto-managing an ephemeral consumer. Both ``name`` and ``deliver_subject``
+    are required by the server.
+    """
+
+    name: str
+    deliver_subject: str
+
+
 @dataclass
 class StreamSource(Base):
     name: str
@@ -191,11 +227,13 @@ class StreamSource(Base):
     filter_subject: Optional[str] = None
     external: Optional[ExternalStream] = None
     subject_transforms: Optional[List[SubjectTransform]] = None
+    consumer: Optional[StreamConsumerSource] = None
 
     @classmethod
     def from_response(cls, resp: Dict[str, Any]):
         cls._convert(resp, "external", ExternalStream)
         cls._convert(resp, "subject_transforms", SubjectTransform)
+        cls._convert(resp, "consumer", StreamConsumerSource)
         cls._convert_utc_iso(resp, "opt_start_time")
         return super().from_response(resp)
 

nats/tests/test_js.py

@@ -4991,6 +4991,58 @@ def test_opt_start_time_non_utc_timezone_preserved_on_parse(self):
         )
 
 
+class StreamConsumerSourceTest(unittest.TestCase):
+    """Unit tests for ADR-60 sourcing-consumer config on StreamSource."""
+
+    def test_stream_source_as_dict_with_consumer(self):
+        src = nats.js.api.StreamSource(
+            name="source-stream",
+            consumer=nats.js.api.StreamConsumerSource(
+                name="durable-consumer",
+                deliver_subject="deliver.subj",
+            ),
+        )
+        d = src.as_dict()
+        assert d["name"] == "source-stream"
+        assert d["consumer"] == {
+            "name": "durable-consumer",
+            "deliver_subject": "deliver.subj",
+        }
+
+    def test_stream_source_as_dict_without_consumer(self):
+        src = nats.js.api.StreamSource(name="source-stream")
+        d = src.as_dict()
+        assert "consumer" not in d
+
+    def test_stream_source_from_response_with_consumer(self):
+        blob = """{
+        "name": "source-stream",
+        "consumer": {"name": "durable-consumer", "deliver_subject": "deliver.subj"}
+        }"""
+        src = nats.js.api.StreamSource.from_response(json.loads(blob))
+        assert src.name == "source-stream"
+        assert isinstance(src.consumer, nats.js.api.StreamConsumerSource)
+        assert src.consumer.name == "durable-consumer"
+        assert src.consumer.deliver_subject == "deliver.subj"
+
+    def test_stream_source_from_response_without_consumer(self):
+        blob = '{"name": "source-stream"}'
+        src = nats.js.api.StreamSource.from_response(json.loads(blob))
+        assert src.consumer is None
+
+    def test_stream_source_consumer_round_trip(self):
+        original = nats.js.api.StreamSource(
+            name="source-stream",
+            consumer=nats.js.api.StreamConsumerSource(
+                name="durable-consumer",
+                deliver_subject="deliver.subj",
+            ),
+        )
+        round_tripped = nats.js.api.StreamSource.from_response(json.loads(json.dumps(original.as_dict())))
+        assert round_tripped.name == original.name
+        assert round_tripped.consumer == original.consumer
+
+
 class V210FeaturesTest(SingleJetStreamServerTestCase):
     @async_test
     async def test_subject_transforms(self):