diff --git a/.gitignore b/.gitignore index a1764b8..d744e63 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,5 @@ /priv erl_crash.dump *.ez -.elixir_ls \ No newline at end of file +.elixir_ls +.claude/ diff --git a/CHANGELOG.md b/CHANGELOG.md index ba253c4..01948c5 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/lib/tcp/client.ex b/lib/tcp/client.ex index 977479f..6746825 100644 --- a/lib/tcp/client.ex +++ b/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(<> = 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 diff --git a/lib/tcp/server/server_handler.ex b/lib/tcp/server/server_handler.ex index 98038d2..a2e4783 100644 --- a/lib/tcp/server/server_handler.ex +++ b/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 diff --git a/lib/tcp/tcp.ex b/lib/tcp/tcp.ex index 47af26d..e32c278 100644 --- a/lib/tcp/tcp.ex +++ b/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(<>, adus) + when byte_size(tail) >= len do + <> = tail + adu = <> + 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(<>, base: :hex)}") + do: + Logger.warning( + "(#{__MODULE__}) Protocol Identifier: #{inspect(<>, base: :hex)}" + ) end diff --git a/test/tcp/modbus_tcp_client_test.exs b/test/tcp/modbus_tcp_client_test.exs index fba0ed7..be95c3e 100644 --- a/test/tcp/modbus_tcp_client_test.exs +++ b/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) + <> = 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 diff --git a/test/tcp/modbus_tcp_server_test.exs b/test/tcp/modbus_tcp_server_test.exs index 09d6a06..a8b7c66 100644 --- a/test/tcp/modbus_tcp_server_test.exs +++ b/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) + <> = 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)