Add async version of send_message for both tcp and rtu

README.rst

@@ -91,7 +91,7 @@ Doing a Modbus request requires even less code:
 
 ..
     Because GitHub doesn't support the include directive the source of
-    scripts/examples/simple_data_store.py has been copied to this file.
+    scripts/examples/simple_tcp_client.py has been copied to this file.
 
 .. code:: python
 
@@ -102,6 +102,8 @@ Doing a Modbus request requires even less code:
 
     from umodbus import conf
     from umodbus.client import tcp
+    from umodbus.client.tcp.asynch import send_message
+
 
     # Enable values to be signed (default is False).
     conf.SIGNED_VALUES = True
@@ -126,6 +128,62 @@ Doing a Modbus request requires even less code:
         response = tcp.send_message(message, sock)
 
 
+uModbus client I/O model is designed to work well with many asynchronous
+concurrency libraries including asyncio_, curio_, trio_, anyio_ and even
+greenlet based libraries like gevent_.
+
+As an example, the above client snippet can be made to work in a gevent
+context simply by replacing the ``import socket`` line with
+``from gevent import socket``.
+
+Here is the same request using any asyncio_ compatible StreamReader and
+StreamWriter objects:
+
+..
+    Because GitHub doesn't support the include directive the source of
+    scripts/examples/simple_tcp_client.py has been copied to this file.
+
+.. code:: python
+
+    #!/usr/bin/env python
+    # scripts/examples/simple_async_tcp_client.py
+    import asyncio
+
+    from umodbus import conf
+    from umodbus.client import tcp
+
+    # Enable values to be signed (default is False).
+    conf.SIGNED_VALUES = True
+
+
+    async def main():
+        # Parse command line arguments
+        parser = ArgumentParser()
+        parser.add_argument("-a", "--address", default="localhost:502")
+
+        args = parser.parse_args()
+        if ":" not in args.address:
+            args.address += ":502"
+        host, port = args.address.rsplit(":", 1)
+        port = int(port)
+
+        reader, writer = await asyncio.open_connection(host, port)
+
+        # Returns a message or Application Data Unit (ADU) specific for doing
+        # Modbus TCP/IP.
+        message = tcp.write_multiple_coils(slave_id=1, starting_address=1, values=[1, 0, 1, 1])
+
+        # Response depends on Modbus function code. This particular returns the
+        # amount of coils written, in this case it is.
+        response = await send_message(message, reader, writer)
+
+        writer.close()
+        await writer.wait_closed()
+
+
+    asyncio.run(main())
+
+
 Features
 --------
 
@@ -156,3 +214,8 @@ Climate Systems`_.
 .. _MODBUS Application Protocol Specification V1.1b3: http://modbus.org/docs/Modbus_Application_Protocol_V1_1b3.pdf
 .. _Mozilla Public License: https://github.com/AdvancedClimateSystems/uModbus/blob/develop/LICENSE
 .. _Read the Docs: http://umodbus.readthedocs.org/en/latest/
+.. _asyncio: https://docs.python.org/3/library/asyncio.html
+.. _curio: https://curio.rtfd.io/
+.. _trio: https://trio.rtfd.io/
+.. _anyio: https://anyio.rtfd.io/
+.. _gevent: http://www.gevent.org/

dev_requirements.txt

@@ -4,6 +4,7 @@ mock==3.0.5;python_version<"3.3"
 pytest==5.3.1;python_version>="3.5"
 pytest==4.6.6;python_version<"3.5"
 pytest-cov==2.8.1
+pytest-asyncio==0.14.0
 Sphinx==2.2.2;python_version>="3.5"
 Sphinx==1.8.5;python_version<"3.5"
 sphinx-rtd-theme==0.4.3

scripts/examples/simple_asyncio_rtu_client.py

@@ -0,0 +1,26 @@
+#!/usr/bin/env python
+# scripts/example/simple_async_rtu_client.py
+import asyncio
+
+from serial_asyncio import open_serial_connection
+
+from umodbus.client.serial import rtu
+from umodbus.client.serial.asynch import send_message
+
+
+async def main():
+    reader, writer = await open_serial_connection(url='/dev/ttyS1', timeout=1)
+
+    # Returns a message or Application Data Unit (ADU) specific for doing
+    # Modbus TCP/IP.
+    message = rtu.write_multiple_coils(slave_id=1, starting_address=1, values=[1, 0, 1, 1])
+
+    # Response depends on Modbus function code. This particular returns the
+    # amount of coils written, in this case it is.
+    response = await send_message(message, reader, writer)
+
+    writer.close()
+    await writer.wait_closed()
+
+
+asyncio.run(main())

scripts/examples/simple_asyncio_tcp_client.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python
+# scripts/examples/simple_async_tcp_client.py
+import asyncio
+from argparse import ArgumentParser
+
+from umodbus import conf
+from umodbus.client import tcp
+from umodbus.client.tcp.asynch import send_message
+
+
+# Enable values to be signed (default is False).
+conf.SIGNED_VALUES = True
+
+
+async def main():
+    # Parse command line arguments
+    parser = ArgumentParser()
+    parser.add_argument("-a", "--address", default="localhost:502")
+
+    args = parser.parse_args()
+    if ":" not in args.address:
+        args.address += ":502"
+    host, port = args.address.rsplit(":", 1)
+    port = int(port)
+
+    reader, writer = await asyncio.open_connection(host, port)
+
+    # Returns a message or Application Data Unit (ADU) specific for doing
+    # Modbus TCP/IP.
+    message = tcp.write_multiple_coils(slave_id=1, starting_address=1, values=[1, 0, 1, 1])
+
+    # Response depends on Modbus function code. This particular returns the
+    # amount of coils written, in this case it is.
+    response = await send_message(message, reader, writer)
+
+    writer.close()
+    await writer.wait_closed()
+
+
+asyncio.run(main())

scripts/examples/simple_curio_tcp_client.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python
+# scripts/examples/simple_curio_tcp_client.py
+import curio
+from argparse import ArgumentParser
+
+from umodbus import conf
+from umodbus.client import tcp
+from umodbus.client.tcp.asynch import send_message
+
+
+# Enable values to be signed (default is False).
+conf.SIGNED_VALUES = True
+
+
+class CurioStream:
+    """Adapter from curio socket to StreamReader/Writer"""
+
+    def __init__(self, sock):
+        self.stream = sock.as_stream()
+        self.data = None
+
+    def write(self, data):
+        self._data = data
+
+    async def drain(self):
+        if self._data:
+            await self.stream.write(self._data)
+            self._data = None
+
+    async def readexactly(self, n):
+        return await self.stream.read_exactly(n)
+
+
+async def main():
+    # Parse command line arguments
+    parser = ArgumentParser()
+    parser.add_argument("-a", "--address", default="localhost:502")
+
+    args = parser.parse_args()
+    if ":" not in args.address:
+        args.address += ":502"
+    host, port = args.address.rsplit(":", 1)
+    port = int(port)
+
+    sock = await curio.open_connection(host, port)
+    stream = CurioStream(sock)
+
+    # Returns a message or Application Data Unit (ADU) specific for doing
+    # Modbus TCP/IP.
+    message = tcp.write_multiple_coils(slave_id=1, starting_address=1, values=[1, 0, 1, 1])
+
+    # Response depends on Modbus function code. This particular returns the
+    # amount of coils written, in this case it is.
+    response = await send_message(message, stream, stream)
+
+    await sock.close()
+
+
+curio.run(main)

setup.py

@@ -22,6 +22,7 @@
       packages=[
           'umodbus',
           'umodbus.client',
+          'umodbus.client.tcp',
           'umodbus.client.serial',
           'umodbus.server',
           'umodbus.server.serial',

tests/system/conftest.py

@@ -1,10 +1,11 @@
 import struct
 import pytest
 import socket
+import asyncio
 from threading import Thread
 
 from .tcp_server import app as tcp
-from .rtu_server import app as rtu
+from .rtu_server import app as rtu, StreamReader, StreamWriter
 
 
 @pytest.fixture(autouse=True, scope="session")
@@ -31,11 +32,32 @@ def sock(tcp_server):
     sock.close()
 
 
+@pytest.yield_fixture
+async def async_tcp_streams(tcp_server):
+    host, port = tcp_server.socket.getsockname()
+    reader, writer = await asyncio.open_connection(host, port)
+
+    yield reader, writer
+
+    writer.close()
+    if hasattr(writer, 'wait_closed'):
+        await writer.wait_closed()
+
+
+
 @pytest.fixture
 def rtu_server():
     return rtu
 
 
+@pytest.fixture
+async def async_serial_streams(rtu_server):
+    reader = StreamReader(rtu_server.serial_port)
+    writer = StreamWriter(rtu_server.serial_port)
+    return reader, writer
+
+
+
 @pytest.fixture
 def mbap():
     return struct.pack('>HHHB', 0, 0, 6, 1)

tests/system/responses/test_exception_async_responses.py

@@ -0,0 +1,89 @@
+import pytest
+import struct
+from functools import partial
+
+from ..validators import validate_response_mbap, validate_response_error
+from umodbus.client import tcp
+
+
+pytestmark = pytest.mark.asyncio
+
+
+async def req_rep(adu, reader, writer):
+    writer.write(adu)
+    await writer.drain()
+    return await reader.read(1024)
+
+
+@pytest.mark.parametrize('function_code, quantity', [
+    (1, 0),
+    (2, 0),
+    (3, 0),
+    (4, 0),
+    (1, 0x07D0 + 1),
+    (2, 0x07D0 + 1),
+    (3, 0x007D + 1),
+    (4, 0x007D + 1),
+])
+async def test_request_returning_invalid_data_value_error(async_tcp_streams, mbap, function_code,
+                                                          quantity):
+    """ Validate response PDU of request returning exception response with
+    error code 3.
+    """
+    starting_address = 0
+    adu = mbap + struct.pack('>BHH', function_code, starting_address, quantity)
+
+    resp = await req_rep(adu, *async_tcp_streams)
+
+    validate_response_mbap(mbap, resp)
+    validate_response_error(resp, function_code, 3)
+
+
+@pytest.mark.parametrize('function', [
+    (partial(tcp.read_coils, 1, 9, 2)),
+    (partial(tcp.read_discrete_inputs, 1, 9, 2)),
+    (partial(tcp.read_holding_registers, 1, 9, 2)),
+    (partial(tcp.read_input_registers, 1, 9, 2)),
+    (partial(tcp.write_single_coil, 1, 11, 0)),
+    (partial(tcp.write_single_register, 1, 11, 1337)),
+    (partial(tcp.write_multiple_coils, 1, 9, [1, 1])),
+    (partial(tcp.write_multiple_registers, 1, 9, [1337, 15])),
+])
+async def test_request_returning_invalid_data_address_error(async_tcp_streams, function):
+    """ Validate response PDU of request returning exception response with
+    error code 2.
+    """
+    adu = function()
+
+    mbap = adu[:7]
+    function_code = struct.unpack('>B', adu[7:8])[0]
+
+    resp = await req_rep(adu, *async_tcp_streams)
+
+    validate_response_mbap(mbap, resp)
+    validate_response_error(resp, function_code, 2)
+
+
+@pytest.mark.parametrize('function', [
+    (partial(tcp.read_coils, 1, 666, 1)),
+    (partial(tcp.read_discrete_inputs, 1, 666, 1)),
+    (partial(tcp.read_holding_registers, 1, 666, 1)),
+    (partial(tcp.read_input_registers, 1, 666, 1)),
+    (partial(tcp.write_single_coil, 1, 666, 0)),
+    (partial(tcp.write_single_register, 1, 666, 1337)),
+    (partial(tcp.write_multiple_coils, 1, 666, [1])),
+    (partial(tcp.write_multiple_registers, 1, 666, [1337])),
+])
+async def test_request_returning_server_device_failure_error(async_tcp_streams, function):
+    """ Validate response PDU of request returning exception response with
+    error code 4.
+    """
+    adu = function()
+
+    mbap = adu[:7]
+    function_code = struct.unpack('>B', adu[7:8])[0]
+
+    resp = await req_rep(adu, *async_tcp_streams)
+
+    validate_response_mbap(mbap, resp)
+    validate_response_error(resp, function_code, 4)