Rename reply_to to reply for compatability with nats.aio

caspervonb · caspervonb · commit a3fc895feccc · 2025-10-22T13:37:53.000+02:00
Signed-off-by: Casper Beyer &lt;casper@synadia.com&gt;
diff --git a/nats-client/src/nats/client/__init__.py b/nats-client/src/nats/client/__init__.py
@@ -297,15 +297,13 @@ async def _read_loop(self) -> None:
                         break
 
                     match msg:
-                        case ("MSG", subject, sid, reply_to, payload):
-                            logger.debug(
-                                "<<- MSG %s %s %s %s", subject, sid, reply_to if reply_to else "", len(payload)
-                            )
-                            await self._handle_msg(subject, sid, reply_to, payload)
-                        case ("HMSG", subject, sid, reply_to, headers, payload, status_code, status_description):
-                            logger.debug("<<- HMSG %s %s %s %s %s", subject, sid, reply_to, len(headers), len(payload))
+                        case ("MSG", subject, sid, reply, payload):
+                            logger.debug("<<- MSG %s %s %s %s", subject, sid, reply if reply else "", len(payload))
+                            await self._handle_msg(subject, sid, reply, payload)
+                        case ("HMSG", subject, sid, reply, headers, payload, status_code, status_description):
+                            logger.debug("<<- HMSG %s %s %s %s %s", subject, sid, reply, len(headers), len(payload))
                             await self._handle_hmsg(
-                                subject, sid, reply_to, headers, payload, status_code, status_description
+                                subject, sid, reply, headers, payload, status_code, status_description
                             )
                         case ("PING",):
                             logger.debug("<<- PING")
@@ -401,11 +399,11 @@ async def _write_loop(self) -> None:
                     logger.exception("Error during final flush")
             return
 
-    async def _handle_msg(self, subject: str, sid: str, reply_to: str | None, payload: bytes) -> None:
+    async def _handle_msg(self, subject: str, sid: str, reply: str | None, payload: bytes) -> None:
         """Handle MSG from server."""
         if sid in self._subscriptions:
             subscription = self._subscriptions[sid]
-            msg = Message(subject=subject, data=payload, reply_to=reply_to)
+            msg = Message(subject=subject, data=payload, reply=reply)
 
             for callback in subscription._callbacks:
                 try:
@@ -422,7 +420,7 @@ async def _handle_hmsg(
         self,
         subject: str,
         sid: str,
-        reply_to: str,
+        reply: str,
         headers: dict[str, list[str]],
         payload: bytes,
         status_code: str | None = None,
@@ -438,7 +436,7 @@ async def _handle_hmsg(
             msg = Message(
                 subject=subject,
                 data=payload,
-                reply_to=reply_to,
+                reply=reply,
                 headers=Headers(headers) if headers else None,  # type: ignore[arg-type]
                 status=status,
             )
@@ -699,7 +697,7 @@ async def publish(
         subject: str,
         payload: bytes,
         *,
-        reply_to: str | None = None,
+        reply: str | None = None,
         headers: Headers | dict[str, str | list[str]] | None = None,
     ) -> None:
         """Publish a message to a subject."""
@@ -712,14 +710,14 @@ async def publish(
             command_parts = encode_hpub(
                 subject,
                 payload,
-                reply_to=reply_to,
+                reply=reply,
                 headers=headers_dict,  # type: ignore[arg-type]
             )
         else:
             command_parts = encode_pub(
                 subject,
                 payload,
-                reply_to=reply_to,
+                reply=reply,
             )
 
         message_data = b"".join(command_parts)
@@ -853,7 +851,7 @@ async def request(
 
         sub = await self.subscribe(inbox)
         try:
-            await self.publish(subject, payload, reply_to=inbox, headers=headers)
+            await self.publish(subject, payload, reply=inbox, headers=headers)
 
             try:
                 response = await asyncio.wait_for(sub.next(), timeout)
diff --git a/nats-client/src/nats/client/message.py b/nats-client/src/nats/client/message.py
@@ -135,13 +135,13 @@ class Message:
     Attributes:
         subject: The subject the message was published to
         data: The message payload as bytes
-        reply_to: Optional reply subject for request-reply messaging
+        reply: Optional reply subject for request-reply messaging
         headers: Optional message headers
         status: Optional NATS status information
     """
 
     subject: str
     data: bytes
-    reply_to: str | None = None
+    reply: str | None = None
     headers: Headers | None = None
     status: Status | None = None
diff --git a/nats-client/src/nats/client/protocol/command.py b/nats-client/src/nats/client/protocol/command.py
@@ -30,20 +30,20 @@ def encode_pub(
     subject: str,
     payload: bytes,
     *,
-    reply_to: str | None = None,
+    reply: str | None = None,
 ) -> list[bytes]:
     """Encode PUB command.
 
     Args:
         subject: Subject to publish to
         payload: Message payload
-        reply_to: Optional reply subject
+        reply: Optional reply subject
 
     Returns:
         List of byte strings to write in sequence
     """
     # PUB format: PUB <subject> [reply-to] <#bytes>
-    command = f"PUB {subject} {reply_to} {len(payload)}\r\n" if reply_to else f"PUB {subject} {len(payload)}\r\n"
+    command = f"PUB {subject} {reply} {len(payload)}\r\n" if reply else f"PUB {subject} {len(payload)}\r\n"
 
     return [command.encode(), payload, b"\r\n"]
 
@@ -52,15 +52,15 @@ def encode_hpub(
     subject: str,
     payload: bytes,
     *,
-    reply_to: str | None = None,
+    reply: str | None = None,
     headers: dict[str, str | list[str]],
 ) -> list[bytes]:
     """Encode HPUB command.
 
     Args:
         subject: Subject to publish to
         payload: Message payload
-        reply_to: Optional reply subject
+        reply: Optional reply subject
         headers: Headers to include with the message
 
     Returns:
@@ -75,8 +75,8 @@ def encode_hpub(
     header_data = ("\r\n".join(header_lines) + "\r\n\r\n").encode()
 
     # HPUB format: HPUB <subject> [reply-to] <#header bytes> <#total bytes>
-    if reply_to:
-        command = f"HPUB {subject} {reply_to} {len(header_data)} {len(header_data) + len(payload)}\r\n"
+    if reply:
+        command = f"HPUB {subject} {reply} {len(header_data)} {len(header_data) + len(payload)}\r\n"
     else:
         command = f"HPUB {subject} {len(header_data)} {len(header_data) + len(payload)}\r\n"
 
diff --git a/nats-client/src/nats/client/protocol/message.py b/nats-client/src/nats/client/protocol/message.py
@@ -66,7 +66,7 @@ class Msg(NamedTuple):
     op: Literal["MSG"]
     subject: str
     sid: str
-    reply_to: str | None
+    reply: str | None
     payload: bytes
 
 
@@ -76,7 +76,7 @@ class HMsg(NamedTuple):
     op: Literal["HMSG"]
     subject: str
     sid: str
-    reply_to: str | None
+    reply: str | None
     headers: dict[str, list[str]]
     payload: bytes
     status_code: str | None
@@ -200,11 +200,11 @@ async def parse_msg(reader: Reader, args: list[bytes]) -> Msg:
 
     if len(args) == MIN_MSG_ARGS:
         # No reply subject
-        reply_to_bytes = None
+        reply_bytes = None
         size = int(args[2])
     else:
         # With reply subject
-        reply_to_bytes = args[2]
+        reply_bytes = args[2]
         size = int(args[3])
 
     # Check payload size limit
@@ -219,9 +219,9 @@ async def parse_msg(reader: Reader, args: list[bytes]) -> Msg:
     # Only convert to strings at the last moment
     subject = subject_bytes.decode()
     sid = sid_bytes.decode()
-    reply_to = reply_to_bytes.decode() if reply_to_bytes is not None else None
+    reply = reply_bytes.decode() if reply_bytes is not None else None
 
-    return Msg("MSG", subject, sid, reply_to, payload)
+    return Msg("MSG", subject, sid, reply, payload)
 
 
 async def parse_hmsg(reader: Reader, args: list[bytes]) -> HMsg:
@@ -247,12 +247,12 @@ async def parse_hmsg(reader: Reader, args: list[bytes]) -> HMsg:
 
     if len(args) == MIN_HMSG_ARGS:
         # No reply subject
-        reply_to_bytes = None
+        reply_bytes = None
         header_size = int(args[2])
         total_size = int(args[3])
     else:
         # With reply subject
-        reply_to_bytes = args[2]
+        reply_bytes = args[2]
         header_size = int(args[3])
         total_size = int(args[4])
 
@@ -281,9 +281,9 @@ async def parse_hmsg(reader: Reader, args: list[bytes]) -> HMsg:
     # Convert remaining bytes to strings only at the final step
     subject = subject_bytes.decode()
     sid = sid_bytes.decode()
-    reply_to = reply_to_bytes.decode() if reply_to_bytes is not None else None
+    reply = reply_bytes.decode() if reply_bytes is not None else None
 
-    return HMsg("HMSG", subject, sid, reply_to, headers, payload, status_code, status_description)
+    return HMsg("HMSG", subject, sid, reply, headers, payload, status_code, status_description)
 
 
 async def parse_info(args: list[bytes]) -> Info:
diff --git a/nats-client/tests/test_client.py b/nats-client/tests/test_client.py
@@ -86,7 +86,7 @@ async def handle_request():
         subscription = await client.subscribe(test_subject)
         await client.flush()
         message = await subscription.next(timeout=1.0)
-        await client.publish(message.reply_to, reply_payload)
+        await client.publish(message.reply, reply_payload)
 
     responder_task = asyncio.create_task(handle_request())
     await client.flush()
@@ -587,18 +587,18 @@ async def test_custom_inbox_prefix(server):
         reply_payload = b"Reply data"
 
         # Track the inbox subject used in the request
-        received_reply_to = None
+        received_reply = None
 
         # Setup responder that captures the reply-to subject
         subscription = await client.subscribe(test_subject)
         await client.flush()
 
         async def handle_request():
-            nonlocal received_reply_to
+            nonlocal received_reply
             message = await subscription.next(timeout=2.0)
-            received_reply_to = message.reply_to
-            assert received_reply_to is not None
-            await client.publish(received_reply_to, reply_payload)
+            received_reply = message.reply
+            assert received_reply is not None
+            await client.publish(received_reply, reply_payload)
 
         responder_task = asyncio.create_task(handle_request())
 
@@ -610,9 +610,9 @@ async def handle_request():
         await responder_task
 
         # Verify that the inbox used the custom prefix
-        assert received_reply_to is not None
-        assert received_reply_to.startswith(custom_prefix), (
-            f"Expected inbox to start with '{custom_prefix}', got '{received_reply_to}'"
+        assert received_reply is not None
+        assert received_reply.startswith(custom_prefix), (
+            f"Expected inbox to start with '{custom_prefix}', got '{received_reply}'"
         )
 
     finally:
diff --git a/nats-client/tests/test_protocol.py b/nats-client/tests/test_protocol.py
@@ -33,7 +33,7 @@ async def test_parse_msg():
     msg = await parse_msg(reader, [b"foo.bar", b"1", b"5"])
     assert msg.subject == "foo.bar"
     assert msg.sid == "1"
-    assert msg.reply_to is None
+    assert msg.reply is None
     assert msg.payload == b"hello"
 
     # Test valid MSG with reply
@@ -44,7 +44,7 @@ async def test_parse_msg():
     msg = await parse_msg(reader, [b"foo.bar", b"1", b"reply.to", b"5"])
     assert msg.subject == "foo.bar"
     assert msg.sid == "1"
-    assert msg.reply_to == "reply.to"
+    assert msg.reply == "reply.to"
     assert msg.payload == b"hello"
 
     # Test invalid size
@@ -78,7 +78,7 @@ async def test_parse_hmsg():
     msg = await parse_hmsg(reader, [b"foo.bar", b"1", b"reply.to", str(header_size).encode(), str(total_size).encode()])
     assert msg.subject == "foo.bar"
     assert msg.sid == "1"
-    assert msg.reply_to == "reply.to"
+    assert msg.reply == "reply.to"
     assert msg.payload == b"hello"
     assert msg.headers == {}
 
@@ -171,7 +171,7 @@ def test_encode_pub():
     assert command == [b"PUB foo.bar 5\r\n", b"hello", b"\r\n"]
 
     # Test with reply
-    command = encode_pub("foo.bar", b"hello", reply_to="reply.to")
+    command = encode_pub("foo.bar", b"hello", reply="reply.to")
     assert command == [b"PUB foo.bar reply.to 5\r\n", b"hello", b"\r\n"]
 
 
@@ -189,7 +189,7 @@ def test_encode_hpub():
     assert command[3] == b"\r\n"
 
     # Test with reply
-    command = encode_hpub("foo.bar", payload, reply_to="reply.to", headers=headers)
+    command = encode_hpub("foo.bar", payload, reply="reply.to", headers=headers)
     assert len(command) == 4
     assert command[0].startswith(b"HPUB foo.bar reply.to")
     assert command[1].startswith(b"NATS/1.0\r\n")
