Modbus TCP: handle multiple ADUs per tcp packet and reassemble fragments

.gitignore

@@ -5,4 +5,5 @@
 /priv
 erl_crash.dump
 *.ez
-.elixir_ls
+.elixir_ls
+.claude/

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 0.3.14
+
+* Modbus TCP Client (active mode): handles multiple responses received in a single tcp packet.
+* Modbus TCP Server: handles multiple requests received in a single tcp packet.
+* Modbus TCP Client & Server: reassemble requests/responses split across tcp packets.
+* Modbus TCP Client: pending requests are discarded when the connection is closed.
+
 ## 0.3.13
 
 * Modbus TCP: the protocol identifier other than Modbus (0x00, 0x00) has been relaxed and logged.

lib/tcp/client.ex

@@ -23,7 +23,8 @@ defmodule Modbux.Tcp.Client do
             d_pid: nil,
             msg_len: 0,
             pending_msg: %{},
-            cmd: nil
+            cmd: nil,
+            buffer: <<>>
 
   @type client_option ::
           {:ip, {byte(), byte(), byte(), byte()}}
@@ -316,40 +317,54 @@ defmodule Modbux.Tcp.Client do
     Logger.debug("(#{__MODULE__}, :message_active) response: #{inspect(response)}")
     Logger.debug("(#{__MODULE__}, :message_active) state: #{inspect(state)}")
 
-    h = :binary.at(response, 0)
-    l = :binary.at(response, 1)
-    transid = h * 256 + l
+    # A tcp packet may carry several responses or just a fragment of one,
+    # incomplete bytes are kept in the buffer until the next packet.
+    {responses, buffer} = Tcp.split_adus(state.buffer <> response)
+    new_state = Enum.reduce(responses, %Client{state | buffer: buffer}, &process_response/2)
+    {:noreply, new_state}
+  end
+
+  def handle_info({:tcp_closed, _port}, state) do
+    Logger.info("(#{__MODULE__}, :tcp_close) Server closed the port")
+    new_state = close_socket(state)
+    {:noreply, new_state}
+  end
+
+  def handle_info(msg, state) do
+    Logger.error("(#{__MODULE__}, :random_msg) msg: #{inspect(msg)}")
+    {:noreply, state}
+  end
+
+  defp process_response(<<transid::16, _::binary>> = response, state) do
     Logger.debug("(#{__MODULE__}, :message_active) transid: #{inspect(transid)}")
 
     case Map.fetch(state.pending_msg, transid) do
       :error ->
         Logger.error("(#{__MODULE__}, :message_active) unknown transaction id")
-        {:noreply, state}
+        state
 
       {:ok, cmd} ->
         values = Tcp.parse_res(cmd, response, transid)
         msg = {:modbus_tcp, cmd, values}
         send(state.d_pid, msg)
         new_pending_msg = Map.delete(state.pending_msg, transid)
-        new_state = %Client{state | cmd: nil, msg_len: 0, pending_msg: new_pending_msg}
-        {:noreply, new_state}
+        %Client{state | cmd: nil, msg_len: 0, pending_msg: new_pending_msg}
     end
   end
 
-  def handle_info({:tcp_closed, _port}, state) do
-    Logger.info("(#{__MODULE__}, :tcp_close) Server close the port")
-    new_state = close_socket(state)
-    {:noreply, new_state}
-  end
-
-  def handle_info(msg, state) do
-    Logger.error("(#{__MODULE__}, :random_msg) msg: #{inspect(msg)}")
-    {:noreply, state}
-  end
-
   defp close_socket(state) do
     :ok = :gen_tcp.close(state.socket)
-    new_state = %Client{state | socket: nil, status: :closed}
+
+    new_state = %Client{
+      state
+      | socket: nil,
+        status: :closed,
+        buffer: <<>>,
+        pending_msg: %{},
+        cmd: nil,
+        msg_len: 0
+    }
+
     new_state
   end
 end

lib/tcp/server/server_handler.ex

@@ -10,7 +10,8 @@ defmodule Modbux.Tcp.Server.Handler do
 
   defstruct model_pid: nil,
             parent_pid: nil,
-            socket: nil
+            socket: nil,
+            buffer: <<>>
 
   @spec start_link([...]) :: :ignore | {:error, any} | {:ok, pid}
   def start_link([socket, model_pid, parent_pid]) do
@@ -23,24 +24,12 @@ defmodule Modbux.Tcp.Server.Handler do
 
   def handle_info({:tcp, socket, data}, state) do
     Logger.debug("(#{__MODULE__}) Received: #{inspect(data, base: :hex)} ")
-    {cmd, transid} = Tcp.parse_req(data)
-    Logger.debug("(#{__MODULE__}) Received Modbux request: #{inspect({cmd, transid})}")
-
-    case Shared.apply(state.model_pid, cmd) do
-      {:ok, values} ->
-        Logger.debug("(#{__MODULE__}) msg send")
-        resp = Tcp.pack_res(cmd, values, transid)
-        if !is_nil(state.parent_pid), do: notify(state.parent_pid, cmd)
-        :ok = :gen_tcp.send(socket, resp)
 
-      {:error, reason} ->
-        Logger.debug("(#{__MODULE__}) An error has occur: #{reason}")
-
-      nil ->
-        Logger.debug("(#{__MODULE__}) Message for another slave")
-    end
-
-    {:noreply, state}
+    # A tcp packet may carry several requests or just a fragment of one,
+    # incomplete bytes are kept in the buffer until the next packet.
+    {requests, buffer} = Tcp.split_adus(state.buffer <> data)
+    Enum.each(requests, fn request -> attend_request(socket, request, state) end)
+    {:noreply, %Handler{state | buffer: buffer}}
   end
 
   def handle_info({:tcp_closed, _socket}, state) do
@@ -67,6 +56,25 @@ defmodule Modbux.Tcp.Server.Handler do
     :gen_tcp.close(state.socket)
   end
 
+  defp attend_request(socket, request, state) do
+    {cmd, transid} = Tcp.parse_req(request)
+    Logger.debug("(#{__MODULE__}) Received Modbux request: #{inspect({cmd, transid})}")
+
+    case Shared.apply(state.model_pid, cmd) do
+      {:ok, values} ->
+        Logger.debug("(#{__MODULE__}) msg send")
+        resp = Tcp.pack_res(cmd, values, transid)
+        if !is_nil(state.parent_pid), do: notify(state.parent_pid, cmd)
+        :ok = :gen_tcp.send(socket, resp)
+
+      {:error, reason} ->
+        Logger.debug("(#{__MODULE__}) An error has occurred: #{reason}")
+
+      nil ->
+        Logger.debug("(#{__MODULE__}) Message for another slave")
+    end
+  end
+
   defp notify(pid, cmd) do
     send(pid, {:modbus_tcp, {:server_request, cmd}})
   end

lib/tcp/tcp.ex

@@ -45,6 +45,22 @@ defmodule Modbux.Tcp do
     Request.length(cmd) + 6
   end
 
+  @doc """
+  Splits a tcp stream into complete Modbus TCP messages (ADUs), each one framed by the
+  length field of its MBAP header. Returns the complete messages and the remaining bytes.
+  """
+  @spec split_adus(binary) :: {[binary], binary}
+  def split_adus(data), do: split_adus(data, [])
+
+  defp split_adus(<<transid::16, protocol_id::16, len::16, tail::binary>>, adus)
+       when byte_size(tail) >= len do
+    <<payload::binary-size(len), rest::binary>> = tail
+    adu = <<transid::16, protocol_id::16, len::16, payload::binary>>
+    split_adus(rest, [adu | adus])
+  end
+
+  defp split_adus(rest, adus), do: {Enum.reverse(adus), rest}
+
   @spec wrap(binary, integer) :: <<_::48, _::_*8>>
   def wrap(payload, transid) do
     size = :erlang.byte_size(payload)
@@ -85,6 +101,10 @@ defmodule Modbux.Tcp do
 
   # Protocol identifier ->	MODBUS protocol = 0x00, 0x00
   defp check_protocol_identifier(0, 0), do: :ok
+
   defp check_protocol_identifier(protocol_id_h, protocol_id_l),
-    do: Logger.warning("(#{__MODULE__}) Protocol Identifier: #{inspect(<<protocol_id_h, protocol_id_l>>, base: :hex)}")
+    do:
+      Logger.warning(
+        "(#{__MODULE__}) Protocol Identifier: #{inspect(<<protocol_id_h, protocol_id_l>>, base: :hex)}"
+      )
 end

test/tcp/modbus_tcp_client_test.exs

@@ -1,5 +1,6 @@
 defmodule ModbuxTcpClientTest do
   use ExUnit.Case
+  alias Modbux.Tcp
   alias Modbux.Tcp.{Client, Server}
 
   test "test Client (connection, stop, configuration)" do
@@ -15,8 +16,78 @@ defmodule ModbuxTcpClientTest do
   end
 
   test "test Client errors" do
-    {:ok, cpid} = Client.start_link(ip: {127, 0, 0, 1}, tcp_port: 5000)
+    {:ok, cpid} = Client.start_link(ip: {127, 0, 0, 1}, tcp_port: 50123)
     assert {:error, :econnrefused} == Client.connect(cpid)
     assert {:error, :closed} == Client.close(cpid)
   end
+
+  # https://github.com/valiot/modbux/issues/14
+  test "client handles two responses in a single tcp packet" do
+    {:ok, listen_socket} = :gen_tcp.listen(0, [:binary, packet: :raw, active: false])
+    {:ok, port} = :inet.port(listen_socket)
+
+    {:ok, cpid} = Client.start_link(ip: {127, 0, 0, 1}, tcp_port: port, active: true)
+    assert :ok == Client.connect(cpid)
+    {:ok, server_socket} = :gen_tcp.accept(listen_socket)
+
+    coils = [1, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1]
+    assert :ok == Client.request(cpid, {:ri, 0, 0, 16})
+    assert :ok == Client.request(cpid, {:fc, 0, 16, 1})
+
+    # Both responses (transid 0 and 1) arrive in the same tcp packet.
+    response = Tcp.pack_res({:ri, 0, 0, 16}, coils, 0) <> Tcp.pack_res({:fc, 0, 16, 1}, nil, 1)
+    :ok = :gen_tcp.send(server_socket, response)
+
+    assert_receive {:modbus_tcp, {:ri, 0, 0, 16}, ^coils}, 1000
+    assert_receive {:modbus_tcp, {:fc, 0, 16, 1}, nil}, 1000
+
+    assert :ok == Client.stop(cpid)
+    :gen_tcp.close(server_socket)
+    :gen_tcp.close(listen_socket)
+  end
+
+  test "client discards pending requests when the socket is closed" do
+    {:ok, listen_socket} = :gen_tcp.listen(0, [:binary, packet: :raw, active: false])
+    {:ok, port} = :inet.port(listen_socket)
+
+    {:ok, cpid} = Client.start_link(ip: {127, 0, 0, 1}, tcp_port: port, active: true)
+    assert :ok == Client.connect(cpid)
+    {:ok, server_socket} = :gen_tcp.accept(listen_socket)
+
+    assert :ok == Client.request(cpid, {:rhr, 0x50, 20817, 1})
+    assert :ok == Client.close(cpid)
+
+    # Requests from a closed connection can never be answered.
+    assert Client.state(cpid).pending_msg == %{}
+
+    assert :ok == Client.stop(cpid)
+    :gen_tcp.close(server_socket)
+    :gen_tcp.close(listen_socket)
+  end
+
+  # https://github.com/valiot/modbux/issues/14
+  test "client reassembles a response split across tcp packets" do
+    # nodelay avoids Nagle coalescing the two fragments into a single tcp packet.
+    {:ok, listen_socket} = :gen_tcp.listen(0, [:binary, packet: :raw, active: false, nodelay: true])
+    {:ok, port} = :inet.port(listen_socket)
+
+    {:ok, cpid} = Client.start_link(ip: {127, 0, 0, 1}, tcp_port: port, active: true)
+    assert :ok == Client.connect(cpid)
+    {:ok, server_socket} = :gen_tcp.accept(listen_socket)
+
+    assert :ok == Client.request(cpid, {:rhr, 0x50, 20817, 2})
+
+    # The response (transid 0) arrives split in two tcp packets.
+    response = Tcp.pack_res({:rhr, 0x50, 20817, 2}, [7, 9], 0)
+    <<first::binary-size(4), second::binary>> = response
+    :ok = :gen_tcp.send(server_socket, first)
+    Process.sleep(50)
+    :ok = :gen_tcp.send(server_socket, second)
+
+    assert_receive {:modbus_tcp, {:rhr, 0x50, 20817, 2}, [7, 9]}, 1000
+
+    assert :ok == Client.stop(cpid)
+    :gen_tcp.close(server_socket)
+    :gen_tcp.close(listen_socket)
+  end
 end

test/tcp/modbus_tcp_server_test.exs

@@ -1,5 +1,6 @@
 defmodule ModbuxTcpServerTest do
   use ExUnit.Case
+  alias Modbux.Tcp
   alias Modbux.Tcp.Server
   alias Modbux.Tcp.Client
 
@@ -47,6 +48,48 @@ defmodule ModbuxTcpServerTest do
     assert_received {:modbus_tcp, {:server_request, {:rc, 80, 20818, 1}}}
   end
 
+  # https://github.com/valiot/modbux/issues/14
+  test "server handles two requests in a single tcp packet" do
+    model = %{0x50 => %{{:c, 20818} => 1, {:hr, 20817} => 7}}
+    {:ok, _spid} = Server.start_link(model: model, port: 2010)
+
+    {:ok, socket} = :gen_tcp.connect({127, 0, 0, 1}, 2010, [:binary, packet: :raw, active: false])
+
+    # Both requests (transid 0 and 1) are sent in the same tcp packet.
+    request = Tcp.pack_req({:rc, 0x50, 20818, 1}, 0) <> Tcp.pack_req({:rhr, 0x50, 20817, 1}, 1)
+    :ok = :gen_tcp.send(socket, request)
+
+    {:ok, resp1} = :gen_tcp.recv(socket, Tcp.res_len({:rc, 0x50, 20818, 1}), 1000)
+    assert Tcp.parse_res({:rc, 0x50, 20818, 1}, resp1, 0) == [1]
+
+    {:ok, resp2} = :gen_tcp.recv(socket, Tcp.res_len({:rhr, 0x50, 20817, 1}), 1000)
+    assert Tcp.parse_res({:rhr, 0x50, 20817, 1}, resp2, 1) == [7]
+
+    :gen_tcp.close(socket)
+  end
+
+  # https://github.com/valiot/modbux/issues/14
+  test "server reassembles a request split across tcp packets" do
+    model = %{0x50 => %{{:hr, 20817} => 7}}
+    {:ok, _spid} = Server.start_link(model: model, port: 2011)
+
+    # nodelay avoids Nagle coalescing the two fragments into a single tcp packet.
+    {:ok, socket} =
+      :gen_tcp.connect({127, 0, 0, 1}, 2011, [:binary, packet: :raw, active: false, nodelay: true])
+
+    # The request (transid 0) is sent split in two tcp packets.
+    request = Tcp.pack_req({:rhr, 0x50, 20817, 1}, 0)
+    <<first::binary-size(5), second::binary>> = request
+    :ok = :gen_tcp.send(socket, first)
+    Process.sleep(50)
+    :ok = :gen_tcp.send(socket, second)
+
+    {:ok, resp} = :gen_tcp.recv(socket, Tcp.res_len({:rhr, 0x50, 20817, 1}), 1000)
+    assert Tcp.parse_res({:rhr, 0x50, 20817, 1}, resp, 0) == [7]
+
+    :gen_tcp.close(socket)
+  end
+
   test "DB updates from Elixir" do
     model = %{0x50 => %{{:c, 0x5152} => 0}}
     {:ok, s_pid} = Server.start_link(model: model, port: 2002, active: true)