remove databuffer from framer.

examples/message_parser.py

@@ -80,7 +80,8 @@ def decode(self, message):
             print(f"{decoder.decoder.__class__.__name__}")
             print("-" * 80)
             try:
-                self.report(decoder.processIncomingFrame(message))
+                _, pdu = decoder.processIncomingFrame(message)
+                self.report(pdu)
             except Exception:  # pylint: disable=broad-except
                 self.check_errors(decoder, message)
 

pymodbus/client/modbusclientprotocol.py

@@ -68,9 +68,10 @@ def callback_data(self, data: bytes, addr: tuple | None = None) -> int:
 
         returns number of bytes consumed
         """
-        if (pdu := self.framer.processIncomingFrame(data)):
+        used_len, pdu = self.framer.processIncomingFrame(data)
+        if pdu:
             self._handle_response(pdu)
-        return len(data)
+        return used_len
 
     def __str__(self):
         """Build a string representation of the connection.

pymodbus/framer/base.py

@@ -65,27 +65,22 @@ def buildFrame(self, message: ModbusPDU) -> bytes:
         frame = self.encode(data, message.slave_id, message.transaction_id)
         return frame
 
-    def processIncomingFrame(self, data: bytes) -> ModbusPDU | None:
+    def processIncomingFrame(self, data: bytes) -> tuple[int, ModbusPDU | None]:
         """Process new packet pattern.
 
         This takes in a new request packet, adds it to the current
         packet stream, and performs framing on it. That is, checks
         for complete messages, and once found, will process all that
         exist.
         """
-        self.databuffer += data
+        used_len = 0
         while True:
-            try:
-                used_len, pdu = self._processIncomingFrame(self.databuffer)
-                if not used_len:
-                    return None
-                if pdu:
-                    self.databuffer = self.databuffer[used_len:]
-                    return pdu
-            except ModbusIOException as exc:
-                self.databuffer = self.EMPTY
-                raise exc
-            self.databuffer = self.databuffer[used_len:]
+            data_len, pdu = self._processIncomingFrame(data[used_len:])
+            used_len += data_len
+            if not data_len:
+                return used_len, None
+            if pdu:
+                return used_len, pdu
 
     def _processIncomingFrame(self, data: bytes) -> tuple[int, ModbusPDU | None]:
         """Process new packet pattern.
@@ -96,15 +91,14 @@ def _processIncomingFrame(self, data: bytes) -> tuple[int, ModbusPDU | None]:
         exist.
         """
         Log.debug("Processing: {}", data, ":hex")
-        while True:
-            if not data:
-                return 0, None
-            used_len, dev_id, tid, frame_data = self.decode(self.databuffer)
-            if not frame_data:
-                return used_len, None
-            if (result := self.decoder.decode(frame_data)) is None:
-                raise ModbusIOException("Unable to decode request")
-            result.slave_id = dev_id
-            result.transaction_id = tid
-            Log.debug("Frame advanced, resetting header!!")
-            return used_len, result
+        if not data:
+            return 0, None
+        used_len, dev_id, tid, frame_data = self.decode(data)
+        if not frame_data:
+            return used_len, None
+        if (result := self.decoder.decode(frame_data)) is None:
+            raise ModbusIOException("Unable to decode request")
+        result.slave_id = dev_id
+        result.transaction_id = tid
+        Log.debug("Frame advanced, resetting header!!")
+        return used_len, result

pymodbus/server/async_io.py

@@ -48,6 +48,7 @@ def __init__(self, owner):
         self.running = False
         self.receive_queue: asyncio.Queue = asyncio.Queue()
         self.handler_task = None  # coroutine to be run on asyncio loop
+        self.databuffer = b''
         self.framer: FramerBase
         self.loop = asyncio.get_running_loop()
 
@@ -117,8 +118,11 @@ async def inner_handle(self):
 
         # if broadcast is enabled make sure to
         # process requests to address 0
-        Log.debug("Handling data: {}", data, ":hex")
-        if (pdu := self.framer.processIncomingFrame(data)):
+        self.databuffer += data
+        Log.debug("Handling data: {}", self.databuffer, ":hex")
+        used_len, pdu = self.framer.processIncomingFrame(self.databuffer)
+        self.databuffer = self.databuffer[used_len:]
+        if pdu:
            self.execute(pdu, *addr)
 
     async def handle(self) -> None:

pymodbus/transaction.py

@@ -137,6 +137,7 @@ def __init__(self, client: ModbusBaseSyncClient, retries):
         self.retries = retries
         self._transaction_lock = RLock()
         self._no_response_devices: list[int] = []
+        self.databuffer = b''
         if client:
             self._set_adu_size()
 
@@ -233,17 +234,20 @@ def execute(self, request: ModbusPDU):  # noqa: C901
                             self._no_response_devices.append(request.slave_id)
                     # No response received and retries not enabled
                     break
-                if (pdu := self.client.framer.processIncomingFrame(response)):
+                self.databuffer += response
+                used_len, pdu = self.client.framer.processIncomingFrame(self.databuffer)
+                self.databuffer = self.databuffer[used_len:]
+                if pdu:
                     self.addTransaction(pdu)
-                if not (response := self.getTransaction(request.transaction_id)):
+                if not (result := self.getTransaction(request.transaction_id)):
                     if len(self.transactions):
-                        response = self.getTransaction(tid=0)
+                        result = self.getTransaction(tid=0)
                     else:
                         last_exception = last_exception or (
                             "No Response received from the remote slave"
                             "/Unable to decode response"
                         )
-                        response = ModbusIOException(
+                        result = ModbusIOException(
                             last_exception, request.function_code
                         )
                     self.client.close()
@@ -254,7 +258,7 @@ def execute(self, request: ModbusPDU):  # noqa: C901
                         '"TRANSACTION_COMPLETE"'
                     )
                     self.client.state = ModbusTransactionState.TRANSACTION_COMPLETE
-                return response
+                return result
             except ModbusIOException as exc:
                 # Handle decode errors method
                 Log.error("Modbus IO exception {}", exc)

test/framers/test_extras.py

@@ -41,46 +41,58 @@ def test_tcp_framer_transaction_half2(self):
         """Test a half completed tcp frame transaction."""
         msg1 = b"\x00\x01\x12\x34\x00\x06\xff"
         msg2 = b"\x02\x01\x02\x00\x08"
-        assert not self._tcp.processIncomingFrame(msg1)
-        result = self._tcp.processIncomingFrame(msg2)
-        assert result
-        assert result.function_code.to_bytes(1,'big') + result.encode() == msg2
+        used_len, pdu = self._tcp.processIncomingFrame(msg1)
+        assert not pdu
+        assert not used_len
+        used_len, pdu = self._tcp.processIncomingFrame(msg1+msg2)
+        assert pdu
+        assert used_len == len(msg1) + len(msg2)
+        assert pdu.function_code.to_bytes(1,'big') + pdu.encode() == msg2
 
     def test_tcp_framer_transaction_half3(self):
         """Test a half completed tcp frame transaction."""
         msg1 = b"\x00\x01\x12\x34\x00\x06\xff\x02\x01\x02\x00"
         msg2 = b"\x08"
-        assert not self._tcp.processIncomingFrame(msg1)
-        result = self._tcp.processIncomingFrame(msg2)
-        assert result
-        assert result.function_code.to_bytes(1,'big') + result.encode() == msg1[7:] + msg2
+        used_len, pdu = self._tcp.processIncomingFrame(msg1)
+        assert not pdu
+        assert not used_len
+        used_len, pdu = self._tcp.processIncomingFrame(msg1+msg2)
+        assert pdu
+        assert used_len == len(msg1) + len(msg2)
+        assert pdu.function_code.to_bytes(1,'big') + pdu.encode() == msg1[7:] + msg2
 
     def test_tcp_framer_transaction_short(self):
         """Test that we can get back on track after an invalid message."""
         msg1 = b''
         msg2 = b"\x00\x01\x12\x34\x00\x06\xff\x02\x01\x02\x00\x08"
-        assert not self._tcp.processIncomingFrame(msg1)
-        result = self._tcp.processIncomingFrame(msg2)
-        assert result
-        assert result.function_code.to_bytes(1,'big') + result.encode() == msg2[7:]
+        used_len, pdu = self._tcp.processIncomingFrame(msg1)
+        assert not pdu
+        assert not used_len
+        used_len, pdu = self._tcp.processIncomingFrame(msg1+msg2)
+        assert pdu
+        assert used_len == len(msg1) + len(msg2)
+        assert pdu.function_code.to_bytes(1,'big') + pdu.encode() == msg2[7:]
 
     def test_tls_incoming_packet(self):
         """Framer tls incoming packet."""
         msg = b"\x00\x01\x12\x34\x00\x06\xff\x02\x12\x34\x01\x02"
-        result = self._tls.processIncomingFrame(msg)
-        assert result
+        _, pdu = self._tls.processIncomingFrame(msg)
+        assert pdu
 
     def test_rtu_process_incoming_packets(self):
         """Test rtu process incoming packets."""
         msg = b"\x00\x01\x00\x00\x00\x01\xfc\x1b"
-        assert self._rtu.processIncomingFrame(msg)
+        _, pdu = self._rtu.processIncomingFrame(msg)
+        assert pdu
 
     def test_ascii_process_incoming_packets(self):
         """Test ascii process incoming packet."""
         msg = b":F7031389000A60\r\n"
-        assert self._ascii.processIncomingFrame(msg)
+        _, pdu = self._ascii.processIncomingFrame(msg)
+        assert pdu
 
     def test_rtu_decode_exception(self):
         """Test that the RTU framer can decode errors."""
         msg = b"\x00\x90\x02\x9c\x01"
-        assert self._rtu.processIncomingFrame(msg)
+        _, pdu = self._rtu.processIncomingFrame(msg)
+        assert pdu

test/framers/test_framer.py

@@ -375,10 +375,11 @@ def test_framer_decode(self, test_framer):
         assert not res_data
 
     @pytest.mark.parametrize(("is_server"), [True])
-    async def x_processIncomingFrame(self, test_framer):
+    async def test_processIncomingFrame1(self, test_framer):
         """Test processIncomingFrame."""
         msg = b"\x00\x01\x00\x00\x00\x01\xfc\x1b"
-        assert test_framer.processIncomingFrame(msg)
+        _, pdu = test_framer.processIncomingFrame(msg)
+        assert pdu
 
     @pytest.mark.parametrize(("is_server"), [True])
     @pytest.mark.parametrize(("entry", "msg"), [
@@ -387,10 +388,11 @@ async def x_processIncomingFrame(self, test_framer):
         (FramerType.RTU, b"\x00\x01\x00\x00\x00\x01\xfc\x1b"),
         (FramerType.ASCII, b":F7031389000A60\r\n"),
     ])
-    def test_processIncomingFrame(self, test_framer, msg):
+    def test_processIncomingFrame2(self, test_framer, msg):
         """Test a tcp frame transaction."""
-        assert test_framer.processIncomingFrame(msg)
-        assert not test_framer.databuffer
+        used_len, pdu = test_framer.processIncomingFrame(msg)
+        assert pdu
+        assert used_len == len(msg)
 
     @pytest.mark.parametrize(("is_server"), [True])
     @pytest.mark.parametrize(("half"), [False, True])
@@ -404,10 +406,13 @@ def test_processIncomingFrame_roundtrip(self, entry, test_framer, msg, dev_id, t
         """Test a tcp frame transaction."""
         if half and entry != FramerType.TLS:
             data_len = int(len(msg) / 2)
-            assert not test_framer.processIncomingFrame(msg[:data_len])
-            result = test_framer.processIncomingFrame(msg[data_len:])
+            used_len, pdu = test_framer.processIncomingFrame(msg[:data_len])
+            assert not pdu
+            assert not used_len
+            used_len, result = test_framer.processIncomingFrame(msg)
         else:
-            result = test_framer.processIncomingFrame(msg)
+            used_len, result = test_framer.processIncomingFrame(msg)
+        assert used_len == len(msg)
         assert result
         assert result.slave_id == dev_id
         assert result.transaction_id == tid