Merge branch 'dev'

Author: janiversen

API_changes.rst

@@ -2,6 +2,12 @@
 PyModbus - API changes.
 =======================
 
+-------------
+Version 3.3.1
+-------------
+
+No changes.
+
 -------------
 Version 3.3.0
 -------------

CHANGELOG.rst

@@ -1,3 +1,16 @@
+version 3.3.1
+----------------------------------------------------------
+* transport fixes and 100% test coverage. (#1580)
+* Delay self.loop until connect(). (#1579)
+* Added mechanism to determine if server did not start cleanly (#1539)
+* Proof transport reconnect works. (#1577)
+* Fix non-shared block doc in config.rst. (#1573)
+
+Thanks to:
+  Hayden Roche
+  jan iversen
+  Philip Couling
+
 version 3.3.0
 ----------------------------------------------------------
 * Stabilize windows tests. (#1567)

MAKE_RELEASE.rst

@@ -8,21 +8,21 @@ Making a release.
 ------------------------------------------------------------
 Prepare/make release on dev.
 ------------------------------------------------------------
-* Make pull request "prepare v3.3.x", with the following:
+* Make pull request "prepare v3.4.x", with the following:
    * Update pymodbus/__init__.py with version number (__version__ X.Y.Zpre)
    * Update README.rst "Supported versions"
    * Update CHANGELOG.rst
       * Add commits from last release, but selectively !
-        git log --oneline v3.2.2..HEAD > commit.log
-        git log v3.2.2..HEAD | grep Author > contributors.log
+        git log --oneline v3.3.0..HEAD > commit.log
+        git log v3.3.0..HEAD | grep Author > contributors.log
    * Commit, push and merge.
 * Checkout master locally
    * git merge dev
    * git push
    * wait for CI to complete on all branches
 * On github "prepare release"
-   * Create tag e.g. v3.0.1dev0
-   * Title "pymodbus v3.0.1dev0"
+   * Create tag e.g. v3.4.0dev0
+   * Title "pymodbus v3.4.0dev0"
    * do NOT generate release notes, but copy from CHANGELOG.rst
    * make release (remember to mark pre-release if so)
 * on local repo
@@ -40,4 +40,4 @@ Prepare release on dev for new commits.
 ------------------------------------------------------------
 * git branch -D master
 * Make pull request "prepare dev", with the following:
-   * Update pymodbus/version.py with version number (last line)
+   * Update pymodbus/__init__.py with version number (__version__ X.Y.Zpre)

README.rst

@@ -22,7 +22,7 @@ Supported versions
 
 Version `2.5.3 <https://github.com/pymodbus-dev/pymodbus/releases/tag/v2.5.3>`_ is the last 2.x release (Supports python >= 2.7, no longer supported).
 
-Version `3.3.0 <https://github.com/pymodbus-dev/pymodbus/releases/tag/v3.3.0>`_ is the current release (Supports Python >= 3.8).
+Version `3.3.0 <https://github.com/pymodbus-dev/pymodbus/releases/tag/v3.3.1>`_ is the current release (Supports Python >= 3.8).
 
 .. important::
    All API changes after 3.0.0 are documented in `API_changes.rst <https://github.com/pymodbus-dev/pymodbus/blob/dev/API_changes.rst>`_

doc/source/library/simulator/config.rst

@@ -258,8 +258,8 @@ Example "setup" configuration:
         assuming all sizes are set to 10, the addresses for configuration are as follows:
             - coils have addresses 0-9,
             - discrete_inputs have addresses 10-19,
-            - holding_registers have addresses 20-29,
-            - input_registers have addresses 30-39
+            - input_registers have addresses 20-29,
+            - holding_registers have addresses 30-39
 
             when configuring the the datatypes (when calling each block start with 0).
 

examples/client_async.py

@@ -121,7 +121,6 @@ async def run_async_client(client, modbus_calls=None):
     """Run sync client."""
     _logger.info("### Client starting")
     await client.connect()
-    print("jan " + str(client.connected))
     assert client.connected
     if modbus_calls:
         await modbus_calls(client)

pymodbus/__init__.py

@@ -12,5 +12,5 @@
 from pymodbus.logging import pymodbus_apply_logging_config
 
 
-__version__ = "3.3.0"
+__version__ = "3.3.1"
 __version_full__ = f"[pymodbus, version {__version__}]"

pymodbus/client/base.py

@@ -14,11 +14,11 @@
 from pymodbus.logging import Log
 from pymodbus.pdu import ModbusRequest, ModbusResponse
 from pymodbus.transaction import DictTransactionManager
-from pymodbus.transport import BaseTransport
+from pymodbus.transport.transport import Transport
 from pymodbus.utilities import ModbusTransactionState
 
 
-class ModbusBaseClient(ModbusClientMixin, BaseTransport):
+class ModbusBaseClient(ModbusClientMixin, Transport):
     """**ModbusBaseClient**
 
     **Parameters common to all clients**:
@@ -94,12 +94,12 @@ def __init__(  # pylint: disable=too-many-arguments
         **kwargs: Any,
     ) -> None:
         """Initialize a client instance."""
-        BaseTransport.__init__(
+        Transport.__init__(
             self,
             "comm",
-            (reconnect_delay * 1000, reconnect_delay_max * 1000),
+            reconnect_delay * 1000,
+            reconnect_delay_max * 1000,
             timeout * 1000,
-            framer,
             lambda: None,
             self.cb_base_connection_lost,
             self.cb_base_handle_data,

pymodbus/server/async_io.py

@@ -534,6 +534,7 @@ def __init__(
 
         # asyncio future that will be done once server has started
         self.serving = asyncio.Future()
+        self.serving_done = asyncio.Future()
         # constructors cannot be declared async, so we have to
         # defer the initialization of the server
         self.server = None
@@ -552,13 +553,15 @@ async def serve_forever(self):
                 Log.info("Server(Unix) listening.")
                 await self.server.serve_forever()
             except asyncio.exceptions.CancelledError:
+                self.serving_done.set_result(True)
                 raise
             except Exception as exc:  # pylint: disable=broad-except
                 Log.error("Server unexpected exception {}", exc)
         else:
             raise RuntimeError(
                 "Can't call serve_forever on an already running server object"
             )
+        self.serving_done.set_result(True)
         Log.info("Server graceful shutdown.")
 
     async def shutdown(self):
@@ -641,6 +644,7 @@ def __init__(
 
         # asyncio future that will be done once server has started
         self.serving = asyncio.Future()
+        self.serving_done = asyncio.Future()
         # constructors cannot be declared async, so we have to
         # defer the initialization of the server
         self.server = None
@@ -663,13 +667,15 @@ async def serve_forever(self):
             try:
                 await self.server.serve_forever()
             except asyncio.exceptions.CancelledError:
+                self.serving_done.set_result(False)
                 raise
             except Exception as exc:  # pylint: disable=broad-except
                 Log.error("Server unexpected exception {}", exc)
         else:
             raise RuntimeError(
                 "Can't call serve_forever on an already running server object"
             )
+        self.serving_done.set_result(True)
         Log.info("Server graceful shutdown.")
 
     async def shutdown(self):
@@ -821,6 +827,7 @@ def __init__(
         self.stop_serving = self.loop.create_future()
         # asyncio future that will be done once server has started
         self.serving = asyncio.Future()
+        self.serving_done = asyncio.Future()
         self.factory_parms = {
             "local_addr": self.address,
             "allow_broadcast": True,
@@ -836,9 +843,11 @@ async def serve_forever(self):
                     **self.factory_parms,
                 )
             except asyncio.exceptions.CancelledError:
+                self.serving_done.set_result(False)
                 raise
             except Exception as exc:
                 Log.error("Server unexpected exception {}", exc)
+                self.serving_done.set_result(False)
                 raise RuntimeError(exc) from exc
             Log.info("Server(UDP) listening.")
             self.serving.set_result(True)
@@ -847,6 +856,7 @@ async def serve_forever(self):
             raise RuntimeError(
                 "Can't call serve_forever on an already running server object"
             )
+        self.serving_done.set_result(True)
 
     async def shutdown(self):
         """Shutdown server."""

pymodbus/transport/__init__.py

@@ -1,7 +1 @@
 """Transport."""
-
-__all__ = [
-    "BaseTransport",
-]
-
-from pymodbus.transport.transport import BaseTransport

pymodbus/transport/transport.py

@@ -10,17 +10,22 @@
 from dataclasses import dataclass
 from typing import Any, Callable, Coroutine
 
-from pymodbus.framer import ModbusFramer
 from pymodbus.logging import Log
 from pymodbus.transport.serial_asyncio import create_serial_connection
 
 
-class BaseTransport:
-    """Base class for transport types.
+class Transport:
+    """Transport layer.
 
-    BaseTransport contains functions common to all transport types and client/server.
+    Contains pure transport methods needed to connect/listen, send/receive and close connections
+    for unix socket, tcp, tls and serial communications.
 
-    This class is not available in the pymodbus API, and should not be referenced in Applications.
+    Contains high level methods like reconnect.
+
+    This class is not available in the pymodbus API, and should not be referenced in Applications
+    nor in the pymodbus documentation.
+
+    The class is designed to be an object in the message level class.
     """
 
     @dataclass
@@ -33,7 +38,6 @@ class CommParamsClass:
         reconnect_delay: float = None
         reconnect_delay_max: float = None
         timeout_connect: float = None
-        framer: ModbusFramer = None
 
         # tcp / tls / udp / serial
         host: str = None
@@ -60,19 +64,19 @@ def check_done(self):
     def __init__(
         self,
         comm_name: str,
-        reconnect_delay: tuple[int, int],
+        reconnect_delay: int,
+        reconnect_max: int,
         timeout_connect: int,
-        framer: ModbusFramer,
         callback_connected: Callable[[], None],
         callback_disconnected: Callable[[Exception], None],
         callback_data: Callable[[bytes], int],
     ) -> None:
         """Initialize a transport instance.
 
         :param comm_name: name of this transport connection
-        :param reconnect_delay: delay and max in milliseconds for first reconnect (0,0 for no reconnect)
+        :param reconnect_delay: delay in milliseconds for first reconnect (0 for no reconnect)
+        :param reconnect_delay: max reconnect delay in milliseconds
         :param timeout_connect: Max. time in milliseconds for connect to complete
-        :param framer: Modbus framer to decode/encode messagees.
         :param callback_connected: Called when connection is established
         :param callback_disconnected: Called when connection is disconnected
         :param callback_data: Called when data is received
@@ -84,25 +88,25 @@ def __init__(
         # properties, can be read, but may not be mingled with
         self.comm_params = self.CommParamsClass(
             comm_name=comm_name,
-            reconnect_delay=reconnect_delay[0] / 1000,
-            reconnect_delay_max=reconnect_delay[1] / 1000,
+            reconnect_delay=reconnect_delay / 1000,
+            reconnect_delay_max=reconnect_max / 1000,
             timeout_connect=timeout_connect / 1000,
-            framer=framer,
         )
 
-        self.reconnect_delay_current: float = 0
+        self.reconnect_delay_current: float = 0.0
         self.transport: asyncio.BaseTransport | asyncio.Server = None
         self.protocol: asyncio.BaseProtocol = None
+        self.loop: asyncio.AbstractEventLoop = None
         with suppress(RuntimeError):
-            self.loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
-        self.reconnect_timer: asyncio.TimerHandle = None
+            self.loop = asyncio.get_running_loop()
+        self.reconnect_task: asyncio.Task = None
         self.recv_buffer: bytes = b""
         self.call_connect_listen: Callable[[], Coroutine[Any, Any, Any]] = lambda: None
         self.use_udp = False
 
-    # ----------------------------- #
-    # Transport specific parameters #
-    # ----------------------------- #
+    # ------------------------ #
+    # Transport specific setup #
+    # ------------------------ #
     def setup_unix(self, setup_server: bool, host: str):
         """Prepare transport unix"""
         if sys.platform.startswith("win"):
@@ -263,6 +267,9 @@ def setup_serial(
     async def transport_connect(self):
         """Handle generic connect and call on to specific transport connect."""
         Log.debug("Connecting {}", self.comm_params.comm_name)
+        if not self.loop:
+            self.loop = asyncio.get_running_loop()
+        self.transport, self.protocol = None, None
         try:
             self.transport, self.protocol = await asyncio.wait_for(
                 self.call_connect_listen(),
@@ -295,6 +302,8 @@ def connection_made(self, transport: asyncio.BaseTransport):
         :param transport: socket etc. representing the connection.
         """
         Log.debug("Connected to {}", self.comm_params.comm_name)
+        if not self.loop:
+            self.loop = asyncio.get_running_loop()
         self.transport = transport
         self.reset_delay()
         self.cb_connection_made()
@@ -306,7 +315,9 @@ def connection_lost(self, reason: Exception):
         """
         Log.debug("Connection lost {} due to {}", self.comm_params.comm_name, reason)
         self.cb_connection_lost(reason)
-        self.close(reconnect=True)
+        if self.transport:
+            self.close()
+            self.reconnect_task = asyncio.create_task(self.reconnect_connect())
 
     def eof_received(self):
         """Call when eof received (other end closed connection).
@@ -352,29 +363,11 @@ def close(self, reconnect: bool = False) -> None:
             self.transport.close()
             self.transport = None
         self.protocol = None
-        if self.reconnect_timer:
-            self.reconnect_timer.cancel()
-            self.reconnect_timer = None
+        if not reconnect and self.reconnect_task:
+            self.reconnect_task.cancel()
+            self.reconnect_task = None
         self.recv_buffer = b""
 
-        if not reconnect or not self.reconnect_delay_current:
-            self.reconnect_delay_current = 0
-            return
-
-        Log.debug(
-            "Waiting {} {} ms reconnecting.",
-            self.comm_params.comm_name,
-            self.reconnect_delay_current * 1000,
-        )
-        self.reconnect_timer = self.loop.call_later(
-            self.reconnect_delay_current,
-            asyncio.create_task,
-            self.transport_connect(),
-        )
-        self.reconnect_delay_current = min(
-            2 * self.reconnect_delay_current, self.comm_params.reconnect_delay_max
-        )
-
     def reset_delay(self) -> None:
         """Reset wait time before next reconnect to minimal period."""
         self.reconnect_delay_current = self.comm_params.reconnect_delay
@@ -386,6 +379,27 @@ def handle_listen(self):
         """Handle incoming connect."""
         return self
 
+    async def reconnect_connect(self):
+        """Handle reconnect as a task."""
+        try:
+            self.reconnect_delay_current = self.comm_params.reconnect_delay
+            transport = None
+            while not transport:
+                Log.debug(
+                    "Wait {} {} ms before reconnecting.",
+                    self.comm_params.comm_name,
+                    self.reconnect_delay_current * 1000,
+                )
+                await asyncio.sleep(self.reconnect_delay_current)
+                transport, _protocol = await self.transport_connect()
+                self.reconnect_delay_current = min(
+                    2 * self.reconnect_delay_current,
+                    self.comm_params.reconnect_delay_max,
+                )
+        except asyncio.CancelledError:
+            pass
+        self.reconnect_task = None
+
     # ----------------- #
     # The magic methods #
     # ----------------- #