move reconnect to transport. (#1507)

Author: janiversen

API_changes.rst

@@ -9,6 +9,7 @@ Version 3.3.0
 - general parameter unit= -> slave=
 - move SqlSlaveContext, RedisSlaveContext to examples/contrib (due to lack of maintenance)
 - :code:`BinaryPayloadBuilder.to_string` was renamed to :code:`BinaryPayloadBuilder.encode`
+- on_reconnect_callback for async clients works slightly different
 
 -------------
 Version 3.2.0

pymodbus/client/base.py

@@ -40,7 +40,7 @@ class ModbusBaseClient(ModbusClientMixin, BaseTransport):
         and not repeated with each client.
 
     .. tip::
-        **delay_ms** doubles automatically with each unsuccessful connect, from
+        **reconnect_delay** doubles automatically with each unsuccessful connect, from
         **reconnect_delay** to **reconnect_delay_max**.
         Set `reconnect_delay=0` to avoid automatic reconnection.
 
@@ -94,7 +94,9 @@ def __init__(  # pylint: disable=too-many-arguments
         **kwargs: Any,
     ) -> None:
         """Initialize a client instance."""
-        BaseTransport.__init__(self)
+        BaseTransport.__init__(
+            self, "comm", framer, reconnect_delay, reconnect_delay_max, timeout, timeout
+        )
         self.params = self._params()
         self.params.framer = framer
         self.params.timeout = float(timeout)
@@ -107,13 +109,19 @@ def __init__(  # pylint: disable=too-many-arguments
         self.reconnect_delay_max = int(reconnect_delay_max)
         self.on_reconnect_callback = on_reconnect_callback
         self.params.kwargs = kwargs
+        self.retry_on_empty: int = 0
+        # -> retry read on nothing
+
+        self.slaves: list[int] = []
+        # -> list of acceptable slaves (0 for accept all)
 
         # Common variables.
         self.framer = self.params.framer(ClientDecoder(), self)
         self.transaction = DictTransactionManager(
             self, retries=retries, retry_on_empty=retry_on_empty, **kwargs
         )
-        self.delay_ms = self.params.reconnect_delay
+        self.reconnect_delay = self.params.reconnect_delay
+        self.reconnect_delay_current = self.params.reconnect_delay
         self.use_protocol = False
         self.use_udp = False
         self.state = ModbusTransactionState.IDLE
@@ -138,15 +146,6 @@ def register(self, custom_response_class: ModbusResponse) -> None:
         """
         self.framer.decoder.register(custom_response_class)
 
-    def connect(self):
-        """Connect to the modbus remote host (call **sync/async**).
-
-        :raises ModbusException: Different exceptions, check exception text.
-
-        **Remark** Retries are handled automatically after first successful connect.
-        """
-        raise NotImplementedException
-
     def is_socket_open(self) -> bool:
         """Return whether socket/serial is open or not (call **sync**)."""
         raise NotImplementedException
@@ -161,10 +160,6 @@ def idle_time(self) -> float:
             return 0
         return self.last_frame_end + self.silent_interval
 
-    def reset_delay(self) -> None:
-        """Reset wait time before next reconnect to minimal period (call **sync**)."""
-        self.delay_ms = self.params.reconnect_delay
-
     def execute(self, request: ModbusRequest = None) -> ModbusResponse:
         """Execute request and get response (call **sync/async**).
 
@@ -199,23 +194,10 @@ async def async_execute(self, request=None):
             try:
                 resp = await asyncio.wait_for(req, timeout=self.params.timeout)
             except asyncio.exceptions.TimeoutError:
-                self.connection_lost("trying to send")
+                self.close(reconnect=True)
                 raise
         return resp
 
-    def connection_lost(self, reason):
-        """Call when the connection is lost or closed.
-
-        The argument is either an exception object or None
-        """
-        Log.debug("Client disconnected from modbus server: {}", reason)
-        self.close(reconnect=True)
-        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.
 
@@ -224,9 +206,25 @@ def data_received(self, data):
         Log.debug("recv: {}", data, ":hex")
         self.framer.processIncomingPacket(data, self._handle_response, slave=0)
 
-    def create_future(self):
-        """Help function to create asyncio Future object."""
-        return asyncio.Future()
+    def cb_handle_data(self, _data: bytes) -> int:
+        """Handle received data
+
+        returns number of bytes consumed
+        """
+
+    def cb_connection_made(self) -> None:
+        """Handle new connection"""
+
+    def cb_connection_lost(self, _reason: Exception) -> None:
+        """Handle lost connection"""
+        for tid in list(self.transaction):
+            self.raise_future(
+                self.transaction.getTransaction(tid),
+                ConnectionException("Connection lost during request"),
+            )
+
+    async def connect(self):
+        """Connect to the modbus remote host."""
 
     def raise_future(self, my_future, exc):
         """Set exception of a future if not done."""
@@ -245,49 +243,17 @@ def _handle_response(self, reply, **_kwargs):
 
     def _build_response(self, tid):
         """Return a deferred response for the current request."""
-        my_future = self.create_future()
+        my_future = asyncio.Future()
         if not self.transport:
             self.raise_future(my_future, ConnectionException("Client is not connected"))
         else:
             self.transaction.addTransaction(my_future, tid)
         return my_future
 
-    def close(self, reconnect: bool = False) -> None:
-        """Close connection.
-
-        :param reconnect: (default false), try to reconnect
-        """
-        if self.transport:
-            if hasattr(self.transport, "_sock"):
-                self.transport._sock.close()  # pylint: disable=protected-access
-            self.transport.abort()
-            self.transport.close()
-            self.transport = None
-        if self._reconnect_task:
-            self._reconnect_task.cancel()
-            self._reconnect_task = None
-
-        if not reconnect or not self.delay_ms:
-            self.delay_ms = 0
-            return
-
-        self._reconnect_task = asyncio.create_task(self._reconnect())
-
-    async def _reconnect(self):
-        """Reconnect."""
-        Log.debug("Waiting {} ms before next connection attempt.", self.delay_ms)
-        await asyncio.sleep(self.delay_ms / 1000)
-        self.delay_ms = min(2 * self.delay_ms, self.reconnect_delay_max)
-
-        self._reconnect_task = None
-        if self.on_reconnect_callback:
-            self.on_reconnect_callback()
-        return await self.connect()
-
     # ----------------------------------------------------------------------- #
     # Internal methods
     # ----------------------------------------------------------------------- #
-    def send(self, request):
+    def send(self, request):  # pylint: disable=invalid-overridden-method
         """Send request.
 
         :meta private:

pymodbus/client/serial.py

@@ -78,7 +78,7 @@ def connected(self):
         """Connect internal."""
         return self.transport is not None
 
-    async def connect(self):  # pylint: disable=invalid-overridden-method
+    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")
@@ -101,6 +101,7 @@ async def connect(self):  # pylint: disable=invalid-overridden-method
         except Exception as exc:  # pylint: disable=broad-except
             Log.warning("Failed to connect: {}", exc)
             self.close(reconnect=True)
+        self.reset_delay()
         return self.connected
 
 
@@ -185,7 +186,7 @@ def connected(self):
         """Connect internal."""
         return self.connect()
 
-    def connect(self):
+    def connect(self):  # pylint: disable=invalid-overridden-method
         """Connect to the modbus serial server."""
         if self.socket:
             return True

pymodbus/client/tcp.py

@@ -51,12 +51,11 @@ def __init__(
         self.params.host = host
         self.params.port = port
         self.params.source_address = source_address
-        self.delay_ms = self.params.reconnect_delay
 
-    async def connect(self):  # pylint: disable=invalid-overridden-method
+    async def connect(self):
         """Initiate connection to start client."""
 
-        # if delay_ms was set to 0 by close(), we need to set it back again
+        # 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()
 
@@ -145,9 +144,9 @@ def __init__(
     @property
     def connected(self):
         """Connect internal."""
-        return self.connect()
+        return self.transport is not None
 
-    def connect(self):
+    def connect(self):  # pylint: disable=invalid-overridden-method
         """Connect to the modbus tcp server."""
         if self.socket:
             return True

pymodbus/client/tls.py

@@ -166,7 +166,7 @@ def __init__(
     @property
     def connected(self):
         """Connect internal."""
-        return self.connect()
+        return self.transport is not None
 
     def connect(self):
         """Connect to the modbus tls server."""

pymodbus/client/udp.py

@@ -54,15 +54,13 @@ def __init__(
         self.params.host = host
         self.params.port = port
         self.params.source_address = source_address
-        self.delay_ms = self.params.reconnect_delay
-        self.reset_delay()
 
     @property
     def connected(self):
         """Return true if connected."""
         return self.transport is not None
 
-    async def connect(self):  # pylint: disable=invalid-overridden-method
+    async def connect(self):
         """Start reconnecting asynchronous udp client.
 
         :meta private:
@@ -96,6 +94,7 @@ async def _connect(self):
         except Exception as exc:  # pylint: disable=broad-except
             Log.warning("Failed to connect: {}", exc)
             self.close(reconnect=True)
+        self.reset_delay()
 
 
 class ModbusUdpClient(ModbusBaseClient):
@@ -140,7 +139,7 @@ def __init__(
 
         self.socket = None
 
-    def connect(self):
+    def connect(self):  # pylint: disable=invalid-overridden-method
         """Connect to the modbus tcp server.
 
         :meta private:
@@ -190,9 +189,7 @@ def is_socket_open(self):
 
         :meta private:
         """
-        if self.socket:
-            return True
-        return self.connect()
+        return True
 
     def __str__(self):
         """Build a string representation of the connection."""

pymodbus/server/async_io.py

@@ -386,7 +386,7 @@ class ModbusDisconnectedRequestHandler(
     def __init__(self, owner):
         """Initialize."""
         super().__init__(owner)
-        _future = asyncio.get_running_loop().create_future()
+        _future = asyncio.Future()
         self.server.on_connection_terminated = _future
 
     def connection_lost(self, call_exc):
@@ -533,7 +533,7 @@ def __init__(
             self.control.Identity.update(identity)
 
         # asyncio future that will be done once server has started
-        self.serving = self.loop.create_future()
+        self.serving = asyncio.Future()
         # constructors cannot be declared async, so we have to
         # defer the initialization of the server
         self.server = None
@@ -641,7 +641,7 @@ def __init__(
             self.control.Identity.update(identity)
 
         # asyncio future that will be done once server has started
-        self.serving = self.loop.create_future()
+        self.serving = asyncio.Future()
         # constructors cannot be declared async, so we have to
         # defer the initialization of the server
         self.server = None
@@ -824,7 +824,7 @@ def __init__(
         self.endpoint = None
         self.on_connection_terminated = None
         # asyncio future that will be done once server has started
-        self.serving = self.loop.create_future()
+        self.serving = asyncio.Future()
         self.factory_parms = {
             "local_addr": self.address,
             "allow_broadcast": True,

pymodbus/transport/__init__.py

@@ -4,4 +4,4 @@
     "BaseTransport",
 ]
 
-from pymodbus.transport.base import BaseTransport
+from pymodbus.transport.transport import BaseTransport