Merge main into multiplex-request-reply-over-a-single-subscription

caspervonb · caspervonb · commit eaa53e2e05af · 2026-04-03T23:29:09.000+02:00
diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml
@@ -92,3 +92,24 @@ jobs:
 
             - name: Run type check
               run: uv run ty check nats-core nats-server
+
+    build:
+        runs-on: ubuntu-latest
+        name: build
+        strategy:
+            matrix:
+                package: [nats-py, nats-core, nats-server, nats-jetstream]
+        steps:
+            - name: Check out repository
+              uses: actions/checkout@v5
+
+            - name: Set up Python
+              uses: actions/setup-python@v6
+              with:
+                  python-version: "3.13"
+
+            - name: Install uv
+              uses: astral-sh/setup-uv@v6
+
+            - name: Build package
+              run: uv build --package ${{ matrix.package }}
diff --git a/nats-core/pyproject.toml b/nats-core/pyproject.toml
@@ -1,6 +1,6 @@
 [build-system]
-requires = ["setuptools>=61.0"]
-build-backend = "setuptools.build_meta"
+requires = ["uv_build>=0.9.28,<0.10.0"]
+build-backend = "uv_build"
 
 [project]
 name = "nats-core"
@@ -32,8 +32,8 @@ Documentation = "https://github.com/nats-io/nats.py"
 Issues = "https://github.com/nats-io/nats.py/issues"
 Source = "https://github.com/nats-io/nats.py"
 
-[tool.setuptools.packages.find]
-where = ["src"]
+[tool.uv.build-backend]
+module-name = "nats.client"
 
 [dependency-groups]
 dev = [
diff --git a/nats-core/src/nats/client/__init__.py b/nats-core/src/nats/client/__init__.py
@@ -913,6 +913,25 @@ async def _force_flush(self) -> None:
         self._pending_messages.clear()
         self._pending_bytes = 0
 
+    async def _ping(self) -> None:
+        """Send a PING to the server."""
+        self._pong_waker.clear()
+        logger.debug("->> PING")
+        self._pings_outstanding += 1
+        self._last_ping_sent = asyncio.get_running_loop().time()
+        await self._connection.write(encode_ping())
+
+    async def rtt(self, timeout: float | None = None) -> float:
+        """Calculate the round trip time between the client and server in seconds."""
+        if self._status == ClientStatus.CLOSED:
+            raise ConnectionError("connection is closed")
+
+        loop = asyncio.get_running_loop()
+        start = loop.time()
+        await self._ping()
+        await asyncio.wait_for(self._pong_waker.wait(), timeout=timeout)
+        return loop.time() - start
+
     async def flush(self, timeout: float | None = None) -> None:
         """Flush pending messages with optional timeout."""
         if self._status == ClientStatus.CLOSED:
@@ -922,11 +941,7 @@ async def flush(self, timeout: float | None = None) -> None:
         if self._pending_messages:
             await self._force_flush()
 
-        self._pong_waker.clear()
-        logger.debug("->> PING")
-        self._pings_outstanding += 1
-        self._last_ping_sent = asyncio.get_event_loop().time()
-        await self._connection.write(encode_ping())
+        await self._ping()
         try:
             await asyncio.wait_for(self._pong_waker.wait(), timeout=timeout)
         except asyncio.TimeoutError:
diff --git a/nats-core/tests/test_client.py b/nats-core/tests/test_client.py
@@ -858,6 +858,12 @@ async def responder():
 
 
 @pytest.mark.asyncio
+async def test_rtt(client):
+    """Test that rtt returns a positive round trip time."""
+    rtt = await client.rtt()
+    assert rtt > 0
+
+
 async def test_flush_ensures_message_delivery(client):
     """Test that flush ensures all pending messages are delivered."""
     test_subject = f"test.flush.{uuid.uuid4()}"
diff --git a/nats-jetstream/pyproject.toml b/nats-jetstream/pyproject.toml
@@ -1,6 +1,6 @@
 [build-system]
-requires = ["setuptools>=61.0"]
-build-backend = "setuptools.build_meta"
+requires = ["uv_build>=0.9.28,<0.10.0"]
+build-backend = "uv_build"
 
 [project]
 name = "nats-jetstream"
@@ -10,9 +10,7 @@ readme = "README.md"
 requires-python = ">=3.11"
 license = "MIT"
 keywords = ["nats", "messaging", "jetstream"]
-authors = [
-  { name = "Casper Beyer", email = "casper@synadia.com" },
-]
+authors = [{ name = "Casper Beyer", email = "casper@synadia.com" }]
 classifiers = [
   "Development Status :: 4 - Beta",
   "Programming Language :: Python",
@@ -22,14 +20,13 @@ classifiers = [
   "Programming Language :: Python :: Implementation :: CPython",
   "Programming Language :: Python :: Implementation :: PyPy",
 ]
-dependencies = [
-  "nats-core",
-]
+dependencies = ["nats-core"]
 
 [dependency-groups]
-dev = [
-  "nats-server",
-]
+dev = ["nats-server"]
+
+[tool.uv.build-backend]
+module-name = "nats.jetstream"
 
 [tool.uv.sources]
 nats-core = { workspace = true }
@@ -40,8 +37,6 @@ Documentation = "https://github.com/nats-io/nats.py"
 Issues = "https://github.com/nats-io/nats.py/issues"
 Source = "https://github.com/nats-io/nats.py"
 
-[tool.setuptools.packages.find]
-where = ["src"]
 
 [tool.pytest.ini_options]
 asyncio_default_fixture_loop_scope = "function"
diff --git a/nats-server/pyproject.toml b/nats-server/pyproject.toml
@@ -1,6 +1,6 @@
 [build-system]
-requires = ["setuptools>=61.0"]
-build-backend = "setuptools.build_meta"
+requires = ["uv_build>=0.9.28,<0.10.0"]
+build-backend = "uv_build"
 
 [project]
 name = "nats-server"
@@ -30,8 +30,8 @@ Documentation = "https://github.com/nats-io/nats.py"
 Issues = "https://github.com/nats-io/nats.py/issues"
 Source = "https://github.com/nats-io/nats.py"
 
-[tool.setuptools.packages.find]
-where = ["src"]
+[tool.uv.build-backend]
+module-name = "nats.server"
 
 [tool.pytest.ini_options]
 asyncio_default_fixture_loop_scope = "function"
diff --git a/nats/examples/graceful_shutdown.py b/nats/examples/graceful_shutdown.py
@@ -0,0 +1,37 @@
+import asyncio
+import signal
+
+import nats
+
+
+async def main():
+    # Connect to NATS
+    nc = await nats.connect("nats://127.0.0.1:4222")
+
+    # Use an event to coordinate graceful shutdown
+    shutdown = asyncio.Event()
+
+    loop = asyncio.get_running_loop()
+    for sig in (signal.SIGTERM, signal.SIGINT):
+        loop.add_signal_handler(sig, shutdown.set)
+
+    async def handler(msg):
+        print(f"Received: {msg.data.decode()} on {msg.subject}")
+
+    await nc.subscribe("foo", cb=handler)
+
+    # Publish messages until shutdown is signalled
+    while not shutdown.is_set():
+        await nc.publish("foo", b"bar")
+        try:
+            await asyncio.wait_for(shutdown.wait(), timeout=1)
+        except asyncio.TimeoutError:
+            pass
+
+    # Drain gracefully closes the connection after flushing
+    await nc.drain()
+    print("Shutdown complete")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/nats/pyproject.toml b/nats/pyproject.toml
@@ -1,6 +1,6 @@
 [build-system]
-requires = ["setuptools>=68.0"]
-build-backend = "setuptools.build_meta"
+requires = ["uv_build>=0.9.28,<0.10.0"]
+build-backend = "uv_build"
 
 [project]
 name = "nats-py"
@@ -35,12 +35,6 @@ nkeys = ['nkeys']
 aiohttp = ['aiohttp']
 fast_parse = ['fast-mail-parser']
 
-[tool.setuptools]
-zip-safe = true
-package-dir = {"" = "src"}
+[tool.uv.build-backend]
+module-name = "nats"
 
-[tool.setuptools.packages.find]
-where = ["src"]
-include = ["nats*"]
-exclude = ["scripts", "tests"]
-namespaces = false
diff --git a/nats/src/nats/js/api.py b/nats/src/nats/js/api.py
@@ -117,7 +117,7 @@ def _parse_utc_iso(time_string: str) -> datetime.datetime:
         if "." in s:
             date_part, frac_tz = s.split(".", 1)
             frac, tz = frac_tz.split("+")
-            frac = frac[:6]  # keep only microseconds
+            frac = frac[:6].ljust(6, "0")  # normalize to exactly 6 digits
             s = f"{date_part}.{frac}+{tz}"
         return datetime.datetime.fromisoformat(s).astimezone(datetime.timezone.utc)
 
@@ -446,6 +446,7 @@ class StreamInfo(Base):
     sources: Optional[List[StreamSourceInfo]] = None
     cluster: Optional[ClusterInfo] = None
     did_create: Optional[bool] = None
+    created: Optional[datetime.datetime] = None
 
     @classmethod
     def from_response(cls, resp: Dict[str, Any]):
@@ -454,6 +455,8 @@ def from_response(cls, resp: Dict[str, Any]):
         cls._convert(resp, "mirror", StreamSourceInfo)
         cls._convert(resp, "sources", StreamSourceInfo)
         cls._convert(resp, "cluster", ClusterInfo)
+
+        cls._convert_utc_iso(resp, "created")
         return super().from_response(resp)
 
 
diff --git a/nats/src/nats/js/kv.py b/nats/src/nats/js/kv.py
@@ -522,6 +522,7 @@ async def watch_updates(msg):
 
         watcher._sub = await self._js.subscribe(
             subject,
+            stream=self._stream,
             cb=watch_updates,
             ordered_consumer=True,
             deliver_policy=deliver_policy,
diff --git a/nats/src/nats/js/manager.py b/nats/src/nats/js/manager.py
@@ -131,6 +131,8 @@ async def update_stream(self, config: Optional[api.StreamConfig] = None, **param
     async def delete_stream(self, name: str) -> bool:
         """
         Delete a stream by name.
+
+        :param name: The name of the stream to delete.
         """
         resp = await self._api_request(f"{self._prefix}.STREAM.DELETE.{name}", timeout=self._timeout)
         return resp["success"]
@@ -228,6 +230,12 @@ async def add_consumer(
         return api.ConsumerInfo.from_response(resp)
 
     async def delete_consumer(self, stream: str, consumer: str) -> bool:
+        """
+        Delete a consumer from a given stream.
+
+        :param stream: The name of the stream from which the consumer should be deleted.
+        :param consumer: The name of the consumer to be deleted.
+        """
         resp = await self._api_request(
             f"{self._prefix}.CONSUMER.DELETE.{stream}.{consumer}",
             b"",
@@ -300,6 +308,7 @@ async def consumers_info(self, stream: str, offset: Optional[int] = None) -> Lis
         """
         consumers_info retrieves a list of consumers. Consumers list limit is 256 for more
         consider to use offset
+
         :param stream: stream to get consumers
         :param offset: consumers list offset
         """
@@ -396,12 +405,17 @@ def _lift_msg_to_raw_msg(self, msg) -> api.RawStreamMsg:
             raw_msg.seq = int(seq)
         raw_msg.data = msg.data
         raw_msg.headers = msg.headers
+        if time_string := msg.headers.get("Nats-Time-Stamp"):
+            raw_msg.time = api.Base._parse_utc_iso(time_string)
 
         return raw_msg
 
     async def delete_msg(self, stream_name: str, seq: int) -> bool:
         """
-        delete_msg retrieves a message from a stream based on the sequence ID.
+        Delete a message from a stream based on the sequence ID.
+
+        :param stream_name: The name of the stream from which the message should be deleted.
+        :param seq: The sequence id to delete
         """
         req_subject = f"{self._prefix}.STREAM.MSG.DELETE.{stream_name}"
         req = {"seq": seq}
diff --git a/nats/src/nats/js/object_store.py b/nats/src/nats/js/object_store.py
@@ -455,9 +455,16 @@ async def watch(
         ignore_deletes=False,
         include_history=False,
         meta_only=False,
+        updates_only=False,
     ) -> ObjectWatcher:
         """
         watch for changes in the underlying store and receive meta information updates.
+
+        :param ignore_deletes: Whether to ignore deleted objects in the updates
+        :param include_history: Whether to include historical values
+        :param meta_only: Whether to only receive metadata
+        :param updates_only: Whether to only receive updates after the current state
+        :return: An ObjectWatcher instance
         """
         all_meta = OBJ_ALL_META_PRE_TEMPLATE.format(
             bucket=self._name,
@@ -473,18 +480,22 @@ async def watch_updates(msg):
 
             # When there are no more updates send an empty marker
             # to signal that it is done, this will unblock iterators
-            if (not watcher._init_done) and meta.num_pending == 0:
+            # Only send None marker when not in updates_only mode
+            if (not watcher._init_done) and meta.num_pending == 0 and not updates_only:
                 watcher._init_done = True
                 await watcher._updates.put(None)
 
         try:
             await self._js.get_last_msg(self._stream, all_meta)
         except NotFoundError:
             watcher._init_done = True
-            await watcher._updates.put(None)
+            if not updates_only:
+                await watcher._updates.put(None)
 
         deliver_policy = None
-        if not include_history:
+        if updates_only:
+            deliver_policy = api.DeliverPolicy.NEW
+        elif not include_history:
             deliver_policy = api.DeliverPolicy.LAST_PER_SUBJECT
 
         watcher._sub = await self._js.subscribe(
diff --git a/nats/tests/test_js.py b/nats/tests/test_js.py
diff --git a/uv.lock b/uv.lock
