remove second client instance in async mode. (#1367)

* merge ModbusClientProtocol to ModbusBaseClient.

Author: janiversen

API_changes.rst

@@ -10,6 +10,8 @@ Version 3.2.0
 - `ReturnSlaveNoReponseCountResponse` has been corrected to
   `ReturnSlaveNoResponseCountResponse`
 - Option `--modbus-config` for REPL server renamed to `--modbus-config-path`
+- client.protocol.<something> --> client.<something>
+- client.factory.<something> --> client.<something>
 
 -------------
 Version 3.1.0

examples/client_async.py

@@ -119,7 +119,7 @@ async def run_async_client(client, modbus_calls=None):
     """Run sync client."""
     _logger.info("### Client starting")
     await client.connect()
-    assert client.protocol
+    assert client.connected
     if modbus_calls:
         await modbus_calls(client)
     await client.close()

pymodbus/client/base.py

@@ -63,10 +63,6 @@ def run():
     **Application methods, common to all clients**:
     """
 
-    state = ModbusTransactionState.IDLE
-    last_frame_end: float = 0
-    silent_interval: float = 0
-
     @dataclass
     class _params:  # pylint: disable=too-many-instance-attributes
         """Parameter class."""
@@ -125,13 +121,16 @@ def __init__(
         self.params.kwargs = kwargs
 
         # Common variables.
-        if xframer := kwargs.get("xframer", None):
-            self.framer = xframer
-        else:
-            self.framer = self.params.framer(ClientDecoder(), self)
+        self.framer = self.params.framer(ClientDecoder(), self)
         self.transaction = DictTransactionManager(self, **kwargs)
         self.delay_ms = self.params.reconnect_delay
-        self.use_protocol = hasattr(self, "protocol")
+        self.use_protocol = False
+        self._connected = False
+        self.use_udp = False
+        self.state = ModbusTransactionState.IDLE
+        self.last_frame_end: float = 0
+        self.silent_interval: float = 0
+        self.transport = None
 
         # Initialize  mixin
         super().__init__()
@@ -185,9 +184,9 @@ def execute(self, request: ModbusRequest = None) -> ModbusResponse:
         :raises ConnectionException: Check exception text.
         """
         if self.use_protocol:
-            if not self.protocol:
+            if not self._connected:
                 raise ConnectionException(f"Not connected[{str(self)}]")
-            return self.protocol.execute(request)
+            return self.async_execute(request)
         if not self.connect():
             raise ConnectionException(f"Failed to connect[{str(self)}]")
         return self.transaction.execute(request)
@@ -196,6 +195,115 @@ def close(self) -> None:
         """Close the underlying socket connection (call **sync/async**)."""
         raise NotImplementedException
 
+    # ----------------------------------------------------------------------- #
+    # Merged client methods
+    # ----------------------------------------------------------------------- #
+    def client_made_connection(self, protocol):
+        """Run transport specific connection."""
+
+    def client_lost_connection(self, protocol):
+        """Run transport specific connection lost."""
+
+    def datagram_received(self, data, _addr):
+        """Receive datagram."""
+        self.data_received(data)
+
+    async def async_execute(self, request=None):
+        """Execute requests asynchronously."""
+        request.transaction_id = self.transaction.getNextTID()
+        packet = self.framer.buildPacket(request)
+        Log.debug("send: {}", packet, ":hex")
+        if self.use_udp:
+            self.transport.sendto(packet)
+        else:
+            self.transport.write(packet)
+        req = self._build_response(request.transaction_id)
+        if self.params.broadcast_enable and not request.unit_id:
+            resp = b"Broadcast write sent - no response expected"
+        else:
+            try:
+                resp = await asyncio.wait_for(req, timeout=self.params.timeout)
+            except asyncio.exceptions.TimeoutError:
+                self.connection_lost("trying to send")
+                raise
+        return resp
+
+    def connection_made(self, transport):
+        """Call when a connection is made.
+
+        The transport argument is the transport representing the connection.
+        """
+        self.transport = transport
+        Log.debug("Client connected to modbus server")
+        self._connected = True
+        self.client_made_connection(self)
+
+    def connection_lost(self, reason):
+        """Call when the connection is lost or closed.
+
+        The argument is either an exception object or None
+        """
+        if self.transport:
+            self.transport.abort()
+            if hasattr(self.transport, "_sock"):
+                self.transport._sock.close()  # pylint: disable=protected-access
+            self.transport = None
+        self.client_lost_connection(self)
+        Log.debug("Client disconnected from modbus server: {}", reason)
+        self._connected = False
+        for tid in list(self.transaction):
+            self.raise_future(
+                self.transaction.getTransaction(tid),
+                ConnectionException("Connection lost during request"),
+            )
+
+    def data_received(self, data):
+        """Call when some data is received.
+
+        data is a non-empty bytes object containing the incoming data.
+        """
+        Log.debug("recv: {}", data, ":hex")
+        self.framer.processIncomingPacket(data, self._handle_response, unit=0)
+
+    def create_future(self):
+        """Help function to create asyncio Future object."""
+        return asyncio.Future()
+
+    def raise_future(self, my_future, exc):
+        """Set exception of a future if not done."""
+        if not my_future.done():
+            my_future.set_exception(exc)
+
+    def _handle_response(self, reply, **_kwargs):
+        """Handle the processed response and link to correct deferred."""
+        if reply is not None:
+            tid = reply.transaction_id
+            if handler := self.transaction.getTransaction(tid):
+                if not handler.done():
+                    handler.set_result(reply)
+            else:
+                Log.debug("Unrequested message: {}", reply, ":str")
+
+    def _build_response(self, tid):
+        """Return a deferred response for the current request."""
+        my_future = self.create_future()
+        if not self._connected:
+            self.raise_future(my_future, ConnectionException("Client is not connected"))
+        else:
+            self.transaction.addTransaction(my_future, tid)
+        return my_future
+
+    @property
+    def async_connected(self):
+        """Return connection status."""
+        return self._connected
+
+    async def async_close(self):
+        """Close connection."""
+        if self.transport:
+            self.transport.close()
+        self._connected = False
+
     # ----------------------------------------------------------------------- #
     # Internal methods
     # ----------------------------------------------------------------------- #
@@ -263,153 +371,3 @@ def __str__(self):
         :returns: The string representation
         """
         return f"{self.__class__.__name__} {self.params.host}:{self.params.port}"
-
-
-class ModbusClientProtocol(
-    ModbusBaseClient,
-    asyncio.Protocol,
-    asyncio.DatagramProtocol,
-):
-    """Asyncio specific implementation of asynchronous modbus client protocol."""
-
-    #: Factory that created this instance.
-    factory = None
-    transport = None
-
-    def __init__(
-        self, host="127.0.0.1", port=502, source_address=None, use_udp=False, **kwargs
-    ):
-        """Initialize a Modbus TCP/UDP asynchronous client"""
-        super().__init__(**kwargs)
-        self.use_udp = use_udp
-        self.params.host = host
-        self.params.port = port
-        self.params.source_address = source_address or ("", 0)
-
-        self._connected = False
-
-    def datagram_received(self, data, addr):
-        """Receive datagram."""
-        self._data_received(data)
-
-    async def execute(self, request=None):  # pylint: disable=invalid-overridden-method
-        """Execute requests asynchronously."""
-        req = self._execute(request)
-        if self.params.broadcast_enable and not request.unit_id:
-            resp = b"Broadcast write sent - no response expected"
-        else:
-            try:
-                resp = await asyncio.wait_for(req, timeout=self.params.timeout)
-            except asyncio.exceptions.TimeoutError:
-                self.connection_lost("trying to send")
-                raise
-        return resp
-
-    def connection_made(self, transport):
-        """Call when a connection is made.
-
-        The transport argument is the transport representing the connection.
-        """
-        self.transport = transport
-        self._connection_made()
-
-        if self.factory:
-            self.factory.protocol_made_connection(self)  # pylint: disable=no-member
-
-    async def close(self):  # pylint: disable=invalid-overridden-method
-        """Close connection."""
-        if self.transport:
-            self.transport.close()
-        self._connected = False
-
-    def connection_lost(self, reason):
-        """Call when the connection is lost or closed.
-
-        The argument is either an exception object or None
-        """
-        if self.transport:
-            self.transport.abort()
-            if hasattr(self.transport, "_sock"):
-                self.transport._sock.close()  # pylint: disable=protected-access
-            self.transport = None
-        if self.factory:
-            self.factory.protocol_lost_connection(self)  # pylint: disable=no-member
-        self._connection_lost(reason)
-
-    def data_received(self, data):
-        """Call when some data is received.
-
-        data is a non-empty bytes object containing the incoming data.
-        """
-        self._data_received(data)
-
-    def create_future(self):
-        """Help function to create asyncio Future object."""
-        return asyncio.Future()
-
-    def resolve_future(self, my_future, result):
-        """Resolve the completed future and sets the result."""
-        if not my_future.done():
-            my_future.set_result(result)
-
-    def raise_future(self, my_future, exc):
-        """Set exception of a future if not done."""
-        if not my_future.done():
-            my_future.set_exception(exc)
-
-    def _connection_made(self):
-        """Call upon a successful client connection."""
-        Log.debug("Client connected to modbus server")
-        self._connected = True
-
-    def _connection_lost(self, reason):
-        """Call upon a client disconnect."""
-        Log.debug("Client disconnected from modbus server: {}", reason)
-        self._connected = False
-        for tid in list(self.transaction):
-            self.raise_future(
-                self.transaction.getTransaction(tid),
-                ConnectionException("Connection lost during request"),
-            )
-
-    @property
-    def connected(self):
-        """Return connection status."""
-        return self._connected
-
-    def write_transport(self, packet):
-        """Write transport."""
-        if self.use_udp:
-            return self.transport.sendto(packet)
-        return self.transport.write(packet)
-
-    def _execute(self, request, **kwargs):  # pylint: disable=unused-argument
-        """Start the producer to send the next request to consumer.write(Frame(request))."""
-        request.transaction_id = self.transaction.getNextTID()
-        packet = self.framer.buildPacket(request)
-        Log.debug("send: {}", packet, ":hex")
-        self.write_transport(packet)
-        return self._build_response(request.transaction_id)
-
-    def _data_received(self, data):
-        """Get response, check for valid message, decode result."""
-        Log.debug("recv: {}", data, ":hex")
-        self.framer.processIncomingPacket(data, self._handle_response, unit=0)
-
-    def _handle_response(self, reply, **kwargs):  # pylint: disable=unused-argument
-        """Handle the processed response and link to correct deferred."""
-        if reply is not None:
-            tid = reply.transaction_id
-            if handler := self.transaction.getTransaction(tid):
-                self.resolve_future(handler, reply)
-            else:
-                Log.debug("Unrequested message: {}", reply, ":str")
-
-    def _build_response(self, tid):
-        """Return a deferred response for the current request."""
-        my_future = self.create_future()
-        if not self._connected:
-            self.raise_future(my_future, ConnectionException("Client is not connected"))
-        else:
-            self.transaction.addTransaction(my_future, tid)
-        return my_future