Local Echo Broadcast with Async Clients

pymodbus/transport/transport.py

@@ -312,10 +312,23 @@ def data_received(self, data: bytes):
     def datagram_received(self, data: bytes, addr: tuple):
         """Receive datagram (UDP connections)."""
         if self.comm_params.handle_local_echo and self.sent_buffer:
-            Log.debug("recv skipping (local_echo): {} addr={}", data, ":hex", addr)
-            if self.sent_buffer in data:
-                data, self.sent_buffer = data.replace(self.sent_buffer, b"", 1), b""
+            if data.startswith(self.sent_buffer):
+                Log.debug(
+                    "recv skipping (local_echo): {} addr={}",
+                    self.sent_buffer,
+                    ":hex",
+                    addr,
+                )
+                data = data[len(self.sent_buffer) :]
+                self.sent_buffer = b""
+            elif self.sent_buffer.startswith(data):
+                Log.debug(
+                    "recv skipping (partial local_echo): {} addr={}", data, ":hex", addr
+                )
+                self.sent_buffer = self.sent_buffer[len(data) :]
+                return
             else:
+                Log.debug("did not receive local echo: {} addr={}", data, ":hex", addr)
                 self.sent_buffer = b""
             if not data:
                 return
@@ -378,7 +391,7 @@ def transport_send(self, data: bytes, addr: tuple = None) -> None:
         """
         Log.debug("send: {}", data, ":hex")
         if self.comm_params.handle_local_echo:
-            self.sent_buffer = data
+            self.sent_buffer += data
         if self.comm_params.comm_type == CommType.UDP:
             if addr:
                 self.transport.sendto(data, addr=addr)  # type: ignore[attr-defined]

test/sub_transport/test_basic.py

@@ -155,6 +155,37 @@ async def test_handle_local_echo(self, client):
         client.datagram_received(test_data, ("127.0.0.1", 502))
         assert client.recv_buffer == test_data + test_data
         assert not client.sent_buffer
+        client.recv_buffer = b""
+        client.transport_send(b"partial")
+        client.datagram_received(b"par", ("127.0.0.1", 502))
+        client.datagram_received(b"tialresponse", ("127.0.0.1", 502))
+        assert client.recv_buffer == b"response"
+        assert not client.sent_buffer
+
+    async def test_broadcast_local_echo(self, client):
+        """Test transport_send() with broadcast and echo packets"""
+        client.comm_params.handle_local_echo = True
+        client.transport = mock.Mock()
+        client.recv_buffer = b""
+        client.transport_send(b"broadcast")
+        client.transport_send(b"message")
+        client.datagram_received(b"broadcast", ("127.0.0.1", 502))
+        client.datagram_received(b"messageresponse", ("127.0.0.1", 502))
+        assert client.recv_buffer == b"response"
+        assert not client.sent_buffer
+        client.recv_buffer = b""
+        client.transport_send(b"broadcast")
+        client.transport_send(b"message")
+        client.datagram_received(b"broadcastmessageresponse", ("127.0.0.1", 502))
+        assert client.recv_buffer == b"response"
+        assert not client.sent_buffer
+        client.recv_buffer = b""
+        client.transport_send(b"broadcast")
+        client.transport_send(b"message")
+        client.datagram_received(b"broadcastmessa", ("127.0.0.1", 502))
+        client.datagram_received(b"geresponse", ("127.0.0.1", 502))
+        assert client.recv_buffer == b"response"
+        assert not client.sent_buffer
 
     async def test_transport_close(self, server, dummy_protocol):
         """Test transport_close()."""