Consolidated into #3311 to keep the related Redis and NATS changes in one PR.

docs/docs/en/nats/jetstream/index.md

@@ -40,9 +40,13 @@ Also, **NATS JetStream** has built-in `key-value` (similar to **Redis**) and `ob
 
 **FastStream** provides access to them via `#!python broker.key_value()` and `#!python broker.object_storage()` — see the [Key-Value Storage](key-value.md){.internal-link} and [Object Storage](object.md){.internal-link} pages. To consume from a stream, pass a `JStream` object to the subscriber:
 
+The same `JStream` object can be reused in `broker.subscriber()`, `broker.publish()`, and `broker.request()`. Pass the target subject separately; a stream can contain multiple subjects.
+
 ```python linenums="1" hl_lines="2 8 12 22"
 {! docs_src/nats/js/main.py !}
 ```
 
 !!! tip
-    Using a `JStream` object, **FastStream** tries to create/update the stream with the object settings. To prevent this behavior and *just get already created stream*, please use `#!python JStream(..., declare=False)` option.
+    When a stream is registered with a subscriber or publisher, **FastStream** tries to create or update it at broker startup using its settings. Use `#!python JStream(..., declare=False)` to work with an existing stream without declaring it.
+
+    Passing a `JStream` to `broker.publish()` or `broker.request()` uses only its name. These calls do not register or declare the stream and do not modify the object.

docs/docs/en/nats/publishing/index.md

@@ -16,7 +16,7 @@ However, if you wish to further customize the publishing logic, you should take
 
 ## NATS Publishing
 
-`NatsBroker` also uses the unified `publish` method (from a `publisher` object) to send messages.
+Use `NatsBroker.publish()` to send messages directly.
 
 ```python
 import asyncio
@@ -47,6 +47,8 @@ The `publish` method accepts the following arguments:
 
 ## NatsJS Parameters
 
-* `#!python stream: str | None = None` - validate that the subject is in the stream.
+* `#!python stream: str | JStream | None = None` - stream name or object used to validate that the subject is in the stream.
 * `#!python timeout: float | None = None` - wait for the NATS server response.
 * `#!python schedule: Optional[Schedule] = None` - Schedule to publish message at a specific time.
+
+See the [JetStream example](../jetstream/index.md){.internal-link} for reusing the same `JStream` object in a subscriber and `broker.publish()`.

docs/docs/en/nats/rpc.md

@@ -13,7 +13,8 @@ search:
 Because **NATS** has zero cost for creating new subjects, we can easily set up a new subject consumer just for the one response message. This way, your request message will be published to one topic, and the response message will be consumed from another one (temporary subject), which allows you to use regular **FastStream RPC** syntax in the **NATS** case too.
 
 !!! tip
-    **FastStream RPC** over **NATS** works in both the *NATS-Core* and *NATS-JS* cases as well, but in the *NATS-JS* case, you have to specify the expected `stream` as a publish argument.
+    **FastStream RPC** over **NATS** works with both *NATS-Core* and *NATS-JS*.
+    For JetStream requests, pass the expected stream name or a `JStream` object as the `stream` argument to `broker.request()`. You can reuse the object registered with the subscriber; the target subject is still passed separately.
 
 ## Blocking Request
 

docs/docs_src/nats/js/main.py

@@ -12,11 +12,11 @@
     stream=stream,
     deliver_policy=DeliverPolicy.NEW,
 )
-async def handler(msg: str, logger: Logger):
+async def handler(msg: str, logger: Logger) -> None:
     logger.info(msg)
 
 @app.after_startup
-async def test_send():
+async def test_send() -> None:
     await broker.publish("Hi!", "js-subject")
     # publish with stream verification
-    await broker.publish("Hi!", "js-subject", stream="stream")
+    await broker.publish("Hi!", "js-subject", stream=stream)

faststream/nats/broker/broker.py

@@ -87,7 +87,7 @@
     from faststream.nats.configs.broker import JsInitOptions
     from faststream.nats.helpers import KVBucketDeclarer, OSBucketDeclarer
     from faststream.nats.message import NatsMessage
-    from faststream.nats.schemas import PubAck, Schedule
+    from faststream.nats.schemas import JStream, PubAck, Schedule
     from faststream.specification.schema.extra import Tag, TagDict
 
     class NatsInitKwargs(TypedDict, total=False):
@@ -778,11 +778,24 @@ async def publish(
         headers: dict[str, str] | None = None,
         reply_to: str = "",
         correlation_id: str | None = None,
-        stream: str | None = None,
+        stream: Union[str, "JStream"] = ...,
         timeout: float | None = None,
         schedule: Optional["Schedule"] = None,
     ) -> "PubAck": ...
 
+    @overload
+    async def publish(
+        self,
+        message: "SendableMessage",
+        subject: str,
+        headers: dict[str, str] | None = None,
+        reply_to: str = "",
+        correlation_id: str | None = None,
+        stream: Union[str, "JStream", None] = None,
+        timeout: float | None = None,
+        schedule: Optional["Schedule"] = None,
+    ) -> Optional["PubAck"]: ...
+
     @override
     async def publish(
         self,
@@ -791,7 +804,7 @@ async def publish(
         headers: dict[str, str] | None = None,
         reply_to: str = "",
         correlation_id: str | None = None,
-        stream: str | None = None,
+        stream: Union[str, "JStream", None] = None,
         timeout: float | None = None,
         schedule: Optional["Schedule"] = None,
     ) -> Optional["PubAck"]:
@@ -817,7 +830,8 @@ async def publish(
                 Manual message **correlation_id** setter.
                 **correlation_id** is a useful option to trace messages.
             stream:
-                This option validates that the target subject is in presented stream.
+                JetStream name or JStream object.
+                Validates that the target subject is in the stream.
                 Can be omitted without any effect if you doesn't want PubAck frame.
             timeout:
                 Timeout to send message to NATS.
@@ -841,7 +855,7 @@ async def publish(
         )
 
         result: PubAck | None
-        if stream:
+        if cmd.stream:
             result = await super()._basic_publish(cmd, producer=self.config.js_producer)
         else:
             result = await super()._basic_publish(cmd, producer=self.config.producer)
@@ -854,7 +868,7 @@ async def request(  # type: ignore[override]
         subject: str,
         headers: dict[str, str] | None = None,
         correlation_id: str | None = None,
-        stream: str | None = None,
+        stream: Union[str, "JStream", None] = None,
         timeout: float = 0.5,
     ) -> "NatsMessage":
         """Make a synchronous request to outer subscriber.
@@ -875,7 +889,8 @@ async def request(  # type: ignore[override]
                 Manual message **correlation_id** setter.
                 **correlation_id** is a useful option to trace messages.
             stream:
-                JetStream name. This option is required if your target subscriber listens for events using JetStream.
+                JetStream name or JStream object.
+                Required if the target subscriber uses JetStream.
             timeout:
                 Timeout to send message to NATS.
 
@@ -892,7 +907,9 @@ async def request(  # type: ignore[override]
             _publish_type=PublishType.REQUEST,
         )
 
-        producer = self.config.js_producer if stream is not None else self.config.producer
+        producer = (
+            self.config.js_producer if cmd.stream is not None else self.config.producer
+        )
 
         msg: NatsMessage = await super()._basic_request(cmd, producer=producer)
         return msg

faststream/nats/response.py

@@ -2,6 +2,7 @@
 
 from typing_extensions import override
 
+from faststream.nats.schemas import JStream
 from faststream.response.publish_type import PublishType
 from faststream.response.response import PublishCommand, Response
 
@@ -51,7 +52,7 @@ def __init__(
         correlation_id: str | None = None,
         headers: dict[str, str] | None = None,
         reply_to: str = "",
-        stream: str | None = None,
+        stream: str | JStream | None = None,
         timeout: float = 0.5,
         schedule: Optional["Schedule"] = None,
         _publish_type: PublishType,
@@ -65,7 +66,7 @@ def __init__(
             _publish_type=_publish_type,
         )
 
-        self.stream = stream
+        self.stream = stream.name if isinstance(stream, JStream) else stream
         self.timeout = timeout
         self.schedule = schedule
 

tests/brokers/nats/test_consume.py

@@ -90,29 +90,37 @@ async def handler(msg: Any) -> None:
         assert event2.is_set()
         assert mock.call_count == 2, mock.call_count
 
+    @pytest.mark.parametrize(
+        "use_stream_object", (False, True), ids=("string", "jstream")
+    )
     async def test_consume_js(
-        self, queue: str, stream: JStream, event: asyncio.Event
+        self,
+        queue: str,
+        stream: JStream,
+        event: asyncio.Event,
+        use_stream_object: bool,
     ) -> None:
         consume_broker = self.get_broker()
+        received: list[str] = []
 
-        args, kwargs = self.get_subscriber_params(queue, stream=stream)
-
-        @consume_broker.subscriber(*args, **kwargs)  # type: ignore[untyped-decorator]
-        def subscriber(m: Any) -> None:
+        @consume_broker.subscriber(queue, stream=stream)
+        async def subscriber(m: str) -> None:
+            received.append(m)
             event.set()
 
         async with self.patch_broker(consume_broker) as br:
             await br.start()
 
-            result = await br.publish("hello", queue, stream=stream.name)
-
-            await asyncio.wait(
-                (asyncio.create_task(event.wait()),),
-                timeout=3,
+            result = await br.publish(
+                "hello", queue, stream=stream if use_stream_object else stream.name
             )
 
+            await asyncio.wait_for(event.wait(), timeout=self.timeout)
+
             assert isinstance(result, PubAck), result
-        assert event.is_set()
+            assert result.stream == stream.name
+
+        assert received == ["hello"]
 
     async def test_consume_with_filter(  # type: ignore[override]
         self, queue: str, mock: MagicMock, event: asyncio.Event

tests/brokers/nats/test_requests.py

@@ -4,6 +4,7 @@
 import pytest
 
 from faststream import BaseMiddleware
+from faststream.nats import JStream, NatsBroker
 from tests.brokers.base.requests import RequestsTestcase
 
 from .basic import NatsMemoryTestcaseConfig, NatsTestcaseConfig
@@ -26,15 +27,17 @@ class NatsRequestsTestcase(RequestsTestcase):
     def get_middleware(self, **kwargs: Any) -> Any:
         return Mid
 
-    async def test_broker_stream_request(self, queue: str) -> None:
-        broker = self.get_broker()
-
-        stream_name = f"{queue}st"
-
-        args, kwargs = self.get_subscriber_params(queue, stream=stream_name)
-
-        @broker.subscriber(*args, **kwargs)  # type: ignore[untyped-decorator]
-        async def handler(msg: Any) -> str:
+    @pytest.mark.parametrize(
+        "use_stream_object", (False, True), ids=("string", "jstream")
+    )
+    async def test_broker_stream_request(
+        self, queue: str, use_stream_object: bool
+    ) -> None:
+        broker: NatsBroker = self.get_broker()
+        stream = JStream(f"{queue}st")
+
+        @broker.subscriber(queue, stream=stream)
+        async def handler(msg: None) -> str:
             return "Response"
 
         async with self.patch_broker(broker):
@@ -44,12 +47,11 @@ async def handler(msg: Any) -> str:
                 None,
                 queue,
                 correlation_id="1",
-                stream=stream_name,
+                stream=stream if use_stream_object else stream.name,
                 timeout=self.timeout,
             )
 
-        assert await response.decode() == "Response"
-        assert response.correlation_id == "1"
+        assert (await response.decode(), response.correlation_id) == ("Response", "1")
 
     async def test_publisher_stream_request(self, queue: str) -> None:
         broker = self.get_broker()

tests/brokers/nats/test_test_client.py

@@ -34,17 +34,24 @@ async def m(msg: Any) -> None: ...
                 await m.assert_called_once_with(path={})
 
     @pytest.mark.asyncio()
+    @pytest.mark.parametrize(
+        "use_stream_object", (False, True), ids=("string", "jstream")
+    )
     async def test_stream_publish(
         self,
         queue: str,
+        use_stream_object: bool,
     ) -> None:
         pub_broker = self.get_broker(apply_types=False)
+        stream = JStream(f"{queue}st")
 
-        @pub_broker.subscriber(queue, stream="test")
-        async def m(msg: Any) -> None: ...
+        @pub_broker.subscriber(queue, stream=stream)
+        async def m(msg: str) -> None: ...
 
         async with self.patch_broker(pub_broker) as br:
-            await br.publish("Hi!", queue, stream="test")
+            await br.publish(
+                "Hi!", queue, stream=stream if use_stream_object else stream.name
+            )
             m.mock.assert_called_once_with("Hi!")
 
     @pytest.mark.asyncio()

tests/mypy/nats.py

@@ -6,6 +6,7 @@
 
 from faststream._internal.basic_types import DecodedMessage
 from faststream.nats import (
+    JStream,
     NatsBroker,
     NatsMessage,
     NatsRoute,
@@ -314,11 +315,14 @@ async def handle21() -> None: ...
 NatsBroker(middlewares=[prometheus_middleware])
 
 
-async def check_broker_publish_result_type() -> None:
+async def check_broker_publish_result_type(stream: str | JStream | None) -> None:
     broker = NatsBroker()
 
     assert_type(await broker.publish(None, "test"), None)
+    assert_type(await broker.publish(None, "test", stream=None), None)
     assert_type(await broker.publish(None, "test", stream="stream"), PubAck)
+    assert_type(await broker.publish(None, "test", stream=JStream("stream")), PubAck)
+    assert_type(await broker.publish(None, "test", stream=stream), PubAck | None)
 
 
 async def check_publisher_publish_result_type() -> None:
@@ -330,11 +334,15 @@ async def check_publisher_publish_result_type() -> None:
     assert_type(await publisher.publish(None, "test", stream="stream"), PubAck)
 
 
-async def check_request_response_type() -> None:
+async def check_request_response_type(stream: str | JStream | None) -> None:
     broker = NatsBroker()
 
     broker_response = await broker.request(None, "test")
     assert_type(broker_response, NatsMessage)
+    assert_type(await broker.request(None, "test", stream=None), NatsMessage)
+    assert_type(await broker.request(None, "test", stream="stream"), NatsMessage)
+    assert_type(await broker.request(None, "test", stream=JStream("stream")), NatsMessage)
+    assert_type(await broker.request(None, "test", stream=stream), NatsMessage)
 
     publisher = broker.publisher("test")
     assert_type(await publisher.request(None, "test"), NatsMessage)