Transport integrated in async clients. (#1541)

Author: janiversen

examples/client_async.py

@@ -131,7 +131,21 @@ async def run_async_client(client, modbus_calls=None):
 
 async def helper():
     """Combine the setup and run"""
-    testclient = setup_async_client(description="Run asynchronous client.")
+    args = [
+        "--comm",
+        "udp",
+        "--host",
+        "127.0.0.1",
+        "--port",
+        "5020",
+        "--framer",
+        "socket",
+        "--log",
+        "debug",
+    ]
+    testclient = setup_async_client(
+        description="Run asynchronous client.", cmdline=args
+    )
     await run_async_client(testclient)
 
 

pymodbus/client/base.py

@@ -8,7 +8,7 @@
 
 from pymodbus.client.mixin import ModbusClientMixin
 from pymodbus.constants import Defaults
-from pymodbus.exceptions import ConnectionException, NotImplementedException
+from pymodbus.exceptions import ConnectionException
 from pymodbus.factory import ClientDecoder
 from pymodbus.framer import ModbusFramer
 from pymodbus.logging import Log
@@ -95,8 +95,16 @@ def __init__(  # pylint: disable=too-many-arguments
     ) -> None:
         """Initialize a client instance."""
         BaseTransport.__init__(
-            self, "comm", framer, reconnect_delay, reconnect_delay_max, timeout, timeout
+            self,
+            "comm",
+            (reconnect_delay * 1000, reconnect_delay_max * 1000),
+            timeout * 1000,
+            framer,
+            lambda: None,
+            self.cb_base_connection_lost,
+            self.cb_base_handle_data,
         )
+        self.framer = framer
         self.params = self._params()
         self.params.framer = framer
         self.params.timeout = float(timeout)
@@ -122,12 +130,11 @@ def __init__(  # pylint: disable=too-many-arguments
         )
         self.reconnect_delay = self.params.reconnect_delay
         self.reconnect_delay_current = self.params.reconnect_delay
-        self.use_protocol = False
+        self.use_sync = False
         self.use_udp = False
         self.state = ModbusTransactionState.IDLE
         self.last_frame_end: float = 0
         self.silent_interval: float = 0
-        self._reconnect_task: asyncio.Task = None
 
         # Initialize  mixin
         ModbusClientMixin.__init__(self)
@@ -146,10 +153,6 @@ def register(self, custom_response_class: ModbusResponse) -> None:
         """
         self.framer.decoder.register(custom_response_class)
 
-    def is_socket_open(self) -> bool:
-        """Return whether socket/serial is open or not (call **sync**)."""
-        raise NotImplementedException
-
     def idle_time(self) -> float:
         """Time before initiating next transaction (call **sync**).
 
@@ -167,13 +170,13 @@ def execute(self, request: ModbusRequest = None) -> ModbusResponse:
         :returns: The result of the request execution
         :raises ConnectionException: Check exception text.
         """
-        if self.use_protocol:
-            if not self.transport:
-                raise ConnectionException(f"Not connected[{str(self)}]")
-            return self.async_execute(request)
-        if not self.connect():
-            raise ConnectionException(f"Failed to connect[{str(self)}]")
-        return self.transaction.execute(request)
+        if self.use_sync:
+            if not self.connect():
+                raise ConnectionException(f"Failed to connect[{str(self)}]")
+            return self.transaction.execute(request)
+        if not self.transport:
+            raise ConnectionException(f"Not connected[{str(self)}]")
+        return self.async_execute(request)
 
     # ----------------------------------------------------------------------- #
     # Merged client methods
@@ -198,24 +201,16 @@ async def async_execute(self, request=None):
                 raise
         return resp
 
-    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, slave=0)
-
-    def cb_handle_data(self, _data: bytes) -> int:
+    def cb_base_handle_data(self, data: bytes) -> int:
         """Handle received data
 
         returns number of bytes consumed
         """
+        Log.debug("recv: {}", data, ":hex")
+        self.framer.processIncomingPacket(data, self._handle_response, slave=0)
+        return len(data)
 
-    def cb_connection_made(self) -> None:
-        """Handle new connection"""
-
-    def cb_connection_lost(self, _reason: Exception) -> None:
+    def cb_base_connection_lost(self, _reason: Exception) -> None:
         """Handle lost connection"""
         for tid in list(self.transaction):
             self.raise_future(

pymodbus/client/serial.py

@@ -6,7 +6,6 @@
 from typing import Any, Type
 
 from pymodbus.client.base import ModbusBaseClient
-from pymodbus.client.serial_asyncio import create_serial_connection
 from pymodbus.constants import Defaults
 from pymodbus.exceptions import ConnectionException
 from pymodbus.framer import ModbusFramer
@@ -60,18 +59,15 @@ def __init__(
         **kwargs: Any,
     ) -> None:
         """Initialize Asyncio Modbus Serial Client."""
-        super().__init__(framer=framer, **kwargs)
-        self.use_protocol = True
+        asyncio.Protocol.__init__(self)
+        ModbusBaseClient.__init__(self, framer=framer, **kwargs)
         self.params.port = port
         self.params.baudrate = baudrate
         self.params.bytesize = bytesize
         self.params.parity = parity
         self.params.stopbits = stopbits
         self.params.handle_local_echo = handle_local_echo
-
-    def _create_protocol(self):
-        """Create a protocol instance."""
-        return self
+        self.setup_serial(False, port, baudrate, bytesize, parity, stopbits)
 
     @property
     def connected(self):
@@ -80,29 +76,13 @@ def connected(self):
 
     async def connect(self):
         """Connect Async client."""
-        # get current loop, if there are no loop a RuntimeError will be raised
-        Log.debug("Starting serial connection")
-        try:
-            await asyncio.wait_for(
-                create_serial_connection(
-                    self.loop,
-                    self._create_protocol,
-                    self.params.port,
-                    baudrate=self.params.baudrate,
-                    bytesize=self.params.bytesize,
-                    stopbits=self.params.stopbits,
-                    parity=self.params.parity,
-                    timeout=self.params.timeout,
-                    **self.params.kwargs,
-                ),
-                timeout=self.params.timeout,
-            )
-            Log.info("Connected to {}", self.params.port)
-        except Exception as exc:  # pylint: disable=broad-except
-            Log.warning("Failed to connect: {}", exc)
-            self.close(reconnect=True)
+        # if reconnect_delay_current was set to 0 by close(), we need to set it back again
+        # so this instance will work
         self.reset_delay()
-        return self.connected
+
+        # force reconnect if required:
+        Log.debug("Connecting to {}.", self.params.host)
+        return await self.transport_connect()
 
 
 class ModbusSerialClient(ModbusBaseClient):
@@ -158,6 +138,7 @@ def __init__(
         self.params.stopbits = stopbits
         self.params.handle_local_echo = handle_local_echo
         self.socket = None
+        self.use_sync = True
 
         self.last_frame_end = None
 
@@ -173,13 +154,12 @@ def __init__(
             else 0.05
         )
 
-        if isinstance(self.framer, ModbusRtuFramer):
-            if self.params.baudrate > 19200:
-                self.silent_interval = 1.75 / 1000  # ms
-            else:
-                self.inter_char_timeout = 1.5 * self._t0
-                self.silent_interval = 3.5 * self._t0
-            self.silent_interval = round(self.silent_interval, 6)
+        if self.params.baudrate > 19200:
+            self.silent_interval = 1.75 / 1000  # ms
+        else:
+            self.inter_char_timeout = 1.5 * self._t0
+            self.silent_interval = 3.5 * self._t0
+        self.silent_interval = round(self.silent_interval, 6)
 
     @property
     def connected(self):

pymodbus/client/tcp.py

@@ -46,11 +46,17 @@ def __init__(
         **kwargs: Any,
     ) -> None:
         """Initialize Asyncio Modbus TCP Client."""
-        super().__init__(framer=framer, **kwargs)
-        self.use_protocol = True
+        asyncio.Protocol.__init__(self)
+        ModbusBaseClient.__init__(self, framer=framer, **kwargs)
         self.params.host = host
         self.params.port = port
         self.params.source_address = source_address
+        if "internal_no_setup" in kwargs:
+            return
+        if host.startswith("unix:"):
+            self.setup_unix(False, host[5:])
+        else:
+            self.setup_tcp(False, host, port)
 
     async def connect(self):
         """Initiate connection to start client."""
@@ -61,45 +67,13 @@ async def connect(self):
 
         # force reconnect if required:
         Log.debug("Connecting to {}:{}.", self.params.host, self.params.port)
-        return await self._connect()
+        return await self.transport_connect()
 
     @property
     def connected(self):
         """Return true if connected."""
         return self.transport is not None
 
-    def _create_protocol(self):
-        """Create initialized protocol instance with function."""
-        return self
-
-    async def _connect(self):
-        """Connect."""
-        Log.debug("Connecting.")
-        try:
-            if self.params.host.startswith("unix:"):
-                transport, protocol = await asyncio.wait_for(
-                    self.loop.create_unix_connection(
-                        self._create_protocol, path=self.params.host[5:]
-                    ),
-                    timeout=self.params.timeout,
-                )
-            else:
-                transport, protocol = await asyncio.wait_for(
-                    self.loop.create_connection(
-                        self._create_protocol,
-                        host=self.params.host,
-                        port=self.params.port,
-                    ),
-                    timeout=self.params.timeout,
-                )
-        except Exception as exc:  # pylint: disable=broad-except
-            Log.warning("Failed to connect: {}", exc)
-            self.close(reconnect=True)
-        else:
-            Log.info("Connected to {}:{}.", self.params.host, self.params.port)
-            self.reset_delay()
-            return transport, protocol
-
 
 class ModbusTcpClient(ModbusBaseClient):
     """**ModbusTcpClient**.
@@ -140,6 +114,7 @@ def __init__(
         self.params.port = port
         self.params.source_address = source_address
         self.socket = None
+        self.use_sync = True
 
     @property
     def connected(self):

pymodbus/client/tls.py

@@ -1,5 +1,4 @@
 """Modbus client async TLS communication."""
-import asyncio
 import socket
 import ssl
 from typing import Any, Type
@@ -40,7 +39,7 @@ def sslctx_provider(
     return sslctx
 
 
-class AsyncModbusTlsClient(AsyncModbusTcpClient, asyncio.Protocol):
+class AsyncModbusTlsClient(AsyncModbusTcpClient):
     """**AsyncModbusTlsClient**.
 
     :param host: Host IP address or host name
@@ -74,41 +73,37 @@ def __init__(
         host: str,
         port: int = Defaults.TlsPort,
         framer: Type[ModbusFramer] = ModbusTlsFramer,
-        sslctx: str = None,
+        sslctx: ssl.SSLContext = None,
         certfile: str = None,
         keyfile: str = None,
         password: str = None,
         server_hostname: str = None,
         **kwargs: Any,
     ):
         """Initialize Asyncio Modbus TLS Client."""
-        super().__init__(host, port=port, framer=framer, **kwargs)
+        AsyncModbusTcpClient.__init__(
+            self, host, port=port, framer=framer, internal_no_setup=True, **kwargs
+        )
         self.sslctx = sslctx_provider(sslctx, certfile, keyfile, password)
-        self.params.sslctx = sslctx
         self.params.certfile = certfile
         self.params.keyfile = keyfile
         self.params.password = password
         self.params.server_hostname = server_hostname
-        AsyncModbusTcpClient.__init__(self, host, port=port, framer=framer, **kwargs)
+        self.setup_tls(
+            False, host, port, sslctx, certfile, keyfile, password, server_hostname
+        )
 
-    async def _connect(self):
-        """Connect to server."""
-        Log.debug("Connecting tls.")
-        try:
-            return await self.loop.create_connection(
-                self._create_protocol,
-                self.params.host,
-                self.params.port,
-                ssl=self.sslctx,
-                server_hostname=self.params.server_hostname,
-            )
-        except Exception as exc:  # pylint: disable=broad-except
-            Log.warning("Failed to connect: {}", exc)
-            self.close(reconnect=True)
-            return
-        Log.info("Connected to {}:{}.", self.params.host, self.params.port)
+    async def connect(self):
+        """Initiate connection to start client."""
+
+        # if reconnect_delay_current was set to 0 by close(), we need to set it back again
+        # so this instance will work
         self.reset_delay()
 
+        # force reconnect if required:
+        Log.debug("Connecting to {}:{}.", self.params.host, self.params.port)
+        return await self.transport_connect()
+
 
 class ModbusTlsClient(ModbusTcpClient):
     """**ModbusTlsClient**.