You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by sr...@apache.org on 2026/10/14 13:39:19 UTC
(plc4x) branch develop updated: feat(plc4net): add UDP transport (#2818)
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new 01934c5bfc feat(plc4net): add UDP transport (#2818)
01934c5bfc is described below
commit 01934c5bfc3b60e28af99b1679c18146c29c79f5
Author: TomNewChao <ch...@gmail.com>
AuthorDate: Wed Oct 14 21:39:12 2026 +0800
feat(plc4net): add UDP transport (#2818)
* feat(plc4net): add UDP transport
* fix(plc4net): harden UDP shutdown and configuration
* fix(plc4net): validate UDP transport options
---
plc4net/plc4net.sln | 15 +
plc4net/test/spi-test/spi-test.csproj | 1 +
.../transports/UdpTransportInstanceTests.cs | 128 +++++++
plc4net/transports/udp/UdpTransport.cs | 254 +++++++++++++
.../transports/udp/UdpTransportConfiguration.cs | 50 +++
plc4net/transports/udp/UdpTransportInstance.cs | 412 +++++++++++++++++++++
.../spi-test.csproj => transports/udp/udp.csproj} | 24 +-
7 files changed, 867 insertions(+), 17 deletions(-)
diff --git a/plc4net/plc4net.sln b/plc4net/plc4net.sln
index fed0e16d9e..f8add6a3dd 100644
--- a/plc4net/plc4net.sln
+++ b/plc4net/plc4net.sln
@@ -55,6 +55,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "s7", "drivers\s7\s7.csproj"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "s7-test", "test\s7-test\s7-test.csproj", "{CDD08B55-2373-4F2D-868E-77C59D6B9CD9}"
EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "udp", "transports\udp\udp.csproj", "{4EE32CEF-9D95-4FDA-B112-94B082F55683}"
+EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@@ -221,6 +223,18 @@ Global
{CDD08B55-2373-4F2D-868E-77C59D6B9CD9}.Release|x64.Build.0 = Release|Any CPU
{CDD08B55-2373-4F2D-868E-77C59D6B9CD9}.Release|x86.ActiveCfg = Release|Any CPU
{CDD08B55-2373-4F2D-868E-77C59D6B9CD9}.Release|x86.Build.0 = Release|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Debug|x64.Build.0 = Debug|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Debug|x86.Build.0 = Debug|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Release|Any CPU.Build.0 = Release|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Release|x64.ActiveCfg = Release|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Release|x64.Build.0 = Release|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Release|x86.ActiveCfg = Release|Any CPU
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683}.Release|x86.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@@ -237,5 +251,6 @@ Global
{544C7992-8497-4AD8-A7D1-A60A49B9B71B} = {0C88DD14-F956-CE84-757C-A364CCF449FC}
{D83FA195-4359-4080-94B5-2839A1925DA0} = {74553503-8F02-3961-70FA-B3C80C1C4ED6}
{CDD08B55-2373-4F2D-868E-77C59D6B9CD9} = {74553503-8F02-3961-70FA-B3C80C1C4ED6}
+ {4EE32CEF-9D95-4FDA-B112-94B082F55683} = {6D81D0D1-7A60-1ADD-C689-B1528D5CD9EE}
EndGlobalSection
EndGlobal
diff --git a/plc4net/test/spi-test/spi-test.csproj b/plc4net/test/spi-test/spi-test.csproj
index 469b3f6b81..f6591f1212 100644
--- a/plc4net/test/spi-test/spi-test.csproj
+++ b/plc4net/test/spi-test/spi-test.csproj
@@ -42,6 +42,7 @@
<ProjectReference Include="..\..\transports\test\test.csproj" />
<ProjectReference Include="..\..\transports\serial\serial.csproj" />
<ProjectReference Include="..\..\transports\cotp\cotp.csproj" />
+ <ProjectReference Include="..\..\transports\udp\udp.csproj" />
</ItemGroup>
</Project>
diff --git a/plc4net/test/spi-test/transports/UdpTransportInstanceTests.cs b/plc4net/test/spi-test/transports/UdpTransportInstanceTests.cs
new file mode 100644
index 0000000000..1c281b9f5e
--- /dev/null
+++ b/plc4net/test/spi-test/transports/UdpTransportInstanceTests.cs
@@ -0,0 +1,128 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+using System;
+using System.Diagnostics;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+using org.apache.plc4net.spi.transports;
+using org.apache.plc4net.transports.udp;
+using Xunit;
+
+namespace org.apache.plc4net.spi.test.transports
+{
+ public class UdpTransportInstanceTests
+ {
+ private const int SettleMillis = 2000;
+
+ private static bool WaitFor(Func<bool> condition)
+ {
+ var deadline = Stopwatch.StartNew();
+ while (deadline.ElapsedMilliseconds < SettleMillis)
+ {
+ if (condition()) return true;
+ Thread.Sleep(5);
+ }
+ return condition();
+ }
+
+ private static UdpTransportInstance Connect(UdpClient peer)
+ {
+ var peerAddress = (IPEndPoint)peer.Client.LocalEndPoint!;
+ return new UdpTransportInstance(
+ new IPEndPoint(IPAddress.Loopback, peerAddress.Port), new UdpTransportConfiguration());
+ }
+
+ [Fact]
+ public void Reads_a_datagram_and_notifies_the_listener()
+ {
+ using var peer = new UdpClient(new IPEndPoint(IPAddress.Loopback, 0));
+ using var instance = Connect(peer);
+ using var notified = new ManualResetEventSlim();
+ instance.RegisterDataListener(() => notified.Set());
+ var bytes = new byte[] { 0x01, 0x02, 0x03 };
+
+ peer.Send(bytes, bytes.Length, instance.LocalAddress);
+
+ Assert.True(notified.Wait(SettleMillis), "the data listener never fired");
+ Assert.True(WaitFor(() => instance.GetNumBytesAvailable() == bytes.Length));
+ Assert.Equal(bytes, instance.Read(bytes.Length));
+ }
+
+ [Fact]
+ public void Writes_one_datagram_to_the_connected_peer()
+ {
+ using var peer = new UdpClient(new IPEndPoint(IPAddress.Loopback, 0));
+ using var instance = Connect(peer);
+ peer.Client.ReceiveTimeout = SettleMillis;
+ var bytes = new byte[] { 0xAA, 0xBB };
+
+ instance.Write(bytes);
+
+ var source = new IPEndPoint(IPAddress.Any, 0);
+ Assert.Equal(bytes, peer.Receive(ref source));
+ }
+
+ [Fact]
+ public void Close_is_idempotent_and_releases_the_socket()
+ {
+ using var peer = new UdpClient(new IPEndPoint(IPAddress.Loopback, 0));
+ using var instance = Connect(peer);
+
+ instance.Close();
+ instance.Close();
+
+ Assert.False(instance.IsOpen);
+ Assert.True(WaitFor(() => instance.ResourcesReleased));
+ }
+
+ [Theory]
+ [InlineData("-1")]
+ [InlineData("65536")]
+ [InlineData("not-a-port")]
+ public void Configuration_rejects_an_invalid_local_port(string localPort)
+ {
+ var transport = new UdpTransport();
+
+ var exception = Assert.Throws<TransportException>(() => transport.CreateConfiguration(
+ new System.Collections.Generic.Dictionary<string, string> { ["local-port"] = localPort }));
+
+ Assert.Contains("local-port", exception.Message);
+ }
+
+ [Fact]
+ public void Configuration_accepts_prefixed_options_and_clamps_the_receive_buffer()
+ {
+ var transport = new UdpTransport(3671);
+ var configuration = Assert.IsType<UdpTransportConfiguration>(transport.CreateConfiguration(
+ new System.Collections.Generic.Dictionary<string, string>
+ {
+ ["udp.local-address"] = "127.0.0.1",
+ ["udp.local-port"] = "12345",
+ ["udp.receive-buffer-size"] = "1"
+ }));
+
+ Assert.Equal(3671, configuration.DefaultPort);
+ Assert.Equal("127.0.0.1", configuration.LocalAddress);
+ Assert.Equal(12345, configuration.LocalPort);
+ Assert.Equal(65535, configuration.ReceiveBufferSize);
+ }
+ }
+}
\ No newline at end of file
diff --git a/plc4net/transports/udp/UdpTransport.cs b/plc4net/transports/udp/UdpTransport.cs
new file mode 100644
index 0000000000..a43065ab50
--- /dev/null
+++ b/plc4net/transports/udp/UdpTransport.cs
@@ -0,0 +1,254 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Linq;
+using System.Net;
+using System.Net.Sockets;
+using org.apache.plc4net.spi.transports;
+
+namespace org.apache.plc4net.transports.udp
+{
+ /// <summary>
+ /// The "udp" transport. Datagram-oriented: there is no connect handshake, and a
+ /// driver's codec sees received bytes in arrival order. It is a connected,
+ /// single-peer transport; broadcast and multicast discovery are outside this
+ /// transport's current contract.
+ /// </summary>
+ public class UdpTransport : ITransport
+ {
+ // The ring buffer must hold at least one maximum-size UDP datagram, or a
+ // single oversized frame stalls the receive loop.
+ private const int MinReceiveBufferSize = 65535;
+ private const int MaxReceiveBufferSize = 64 * 1024 * 1024;
+
+ private readonly int _defaultPort;
+
+ public UdpTransport() : this(-1)
+ {
+ }
+
+ /// <param name="defaultPort">
+ /// Port to use when the connection string omits one (e.g. 3671 for KNXnet/IP).
+ /// A <c>default-port</c> connection-string parameter still overrides it.
+ /// </param>
+ public UdpTransport(int defaultPort)
+ {
+ _defaultPort = defaultPort;
+ }
+
+ public string TransportCode => "udp";
+
+ public string TransportName => "UDP Datagram Transport";
+
+ public ITransportConfiguration CreateConfiguration(IReadOnlyDictionary<string, string> parameters)
+ {
+ var configuration = new UdpTransportConfiguration { DefaultPort = _defaultPort };
+ if (parameters == null)
+ {
+ return configuration;
+ }
+
+ // Options may be written bare or prefixed with the transport code, matching
+ // the TCP transport and the Java ConfigurationFactory convention.
+ configuration.DefaultPort = GetInt(parameters, "default-port", configuration.DefaultPort);
+ configuration.LocalAddress = GetString(parameters, "local-address", configuration.LocalAddress);
+ configuration.LocalPort = GetInt(parameters, "local-port", configuration.LocalPort);
+ configuration.ReceiveBufferSize = GetInt(parameters, "receive-buffer-size", configuration.ReceiveBufferSize);
+
+ if (configuration.DefaultPort != -1
+ && (configuration.DefaultPort < 1 || configuration.DefaultPort > 65535))
+ {
+ throw new TransportException(
+ $"default-port must be between 1 and 65535, but was {configuration.DefaultPort}.");
+ }
+ if (configuration.LocalPort < 0 || configuration.LocalPort > 65535)
+ {
+ throw new TransportException(
+ $"local-port must be between 0 and 65535, but was {configuration.LocalPort}.");
+ }
+
+ // Clamp the ring-buffer size: 0/negative would crash the RingBuffer
+ // constructor, and anything below one full datagram lets a single
+ // oversized frame stall the receive loop.
+ configuration.ReceiveBufferSize = Math.Clamp(
+ configuration.ReceiveBufferSize, MinReceiveBufferSize, MaxReceiveBufferSize);
+
+ return configuration;
+ }
+
+ public ITransportInstance CreateTransportInstance(
+ string transportConfig, ITransportConfiguration configuration)
+ {
+ if (!(configuration is UdpTransportConfiguration udpConfiguration))
+ {
+ throw new TransportException(
+ $"Expected a {nameof(UdpTransportConfiguration)} but got {configuration?.GetType().Name ?? "null"}.");
+ }
+
+ var (host, port, driverConfig) = ParseAddress(transportConfig, udpConfiguration.DefaultPort);
+ var endPoint = new IPEndPoint(ResolveHost(host), port);
+
+ var instance = new UdpTransportInstance(endPoint, udpConfiguration);
+ instance.SetDriverConfig(driverConfig);
+ return instance;
+ }
+
+ /// <summary>
+ /// Splits "host:port/driver-config" into its parts. The port may be omitted
+ /// when the driver supplies a default. Bracketed IPv6 literals are supported.
+ /// Shares its rules with the TCP transport.
+ /// </summary>
+ internal static (string Host, int Port, string DriverConfig) ParseAddress(string address, int defaultPort)
+ {
+ if (string.IsNullOrWhiteSpace(address))
+ {
+ throw new TransportException("Transport address must not be empty.");
+ }
+
+ var driverConfig = string.Empty;
+ var slash = address.IndexOf('/');
+ if (slash >= 0)
+ {
+ driverConfig = address.Substring(slash);
+ address = address.Substring(0, slash);
+ }
+
+ string host;
+ int port;
+
+ if (address.StartsWith("[", StringComparison.Ordinal))
+ {
+ var closing = address.IndexOf(']');
+ if (closing < 0)
+ {
+ throw new TransportException($"Malformed IPv6 address '{address}'.");
+ }
+ host = address.Substring(1, closing - 1);
+ var rest = address.Substring(closing + 1);
+ if (rest.Length > 0 && !rest.StartsWith(":", StringComparison.Ordinal))
+ {
+ throw new TransportException(
+ $"Malformed address '{address}': unexpected '{rest}' after ']'.");
+ }
+ port = rest.StartsWith(":", StringComparison.Ordinal)
+ ? ParsePort(rest.Substring(1))
+ : RequireDefaultPort(defaultPort, address);
+ }
+ else
+ {
+ var parts = address.Split(':');
+ if (parts.Length == 1)
+ {
+ host = parts[0];
+ port = RequireDefaultPort(defaultPort, address);
+ }
+ else if (parts.Length == 2)
+ {
+ host = parts[0];
+ port = ParsePort(parts[1]);
+ }
+ else
+ {
+ throw new TransportException(
+ $"Malformed address '{address}'. Use host, host:port, or [ipv6]:port.");
+ }
+ }
+
+ if (string.IsNullOrWhiteSpace(host))
+ {
+ throw new TransportException($"Address '{address}' has no host part.");
+ }
+
+ return (host, port, driverConfig);
+ }
+
+ private static int RequireDefaultPort(int defaultPort, string address)
+ {
+ if (defaultPort <= 0)
+ {
+ throw new TransportException(
+ $"Address '{address}' has no port and the driver declares no default port.");
+ }
+ return defaultPort;
+ }
+
+ private static int ParsePort(string raw)
+ {
+ if (!int.TryParse(raw, NumberStyles.Integer, CultureInfo.InvariantCulture, out var port)
+ || port <= 0 || port > 65535)
+ {
+ throw new TransportException($"'{raw}' is not a valid port number.");
+ }
+ return port;
+ }
+
+ private static IPAddress ResolveHost(string host)
+ {
+ if (IPAddress.TryParse(host, out var address))
+ {
+ return address;
+ }
+ try
+ {
+ var resolved = Dns.GetHostAddresses(host);
+ if (resolved.Length == 0)
+ {
+ throw new TransportException($"Host '{host}' did not resolve to any address.");
+ }
+ // Prefer IPv4: industrial gateways are typically v4-only, and a v6-first
+ // resolver result would build a v6 socket that cannot reach them.
+ return resolved.FirstOrDefault(a => a.AddressFamily == AddressFamily.InterNetwork)
+ ?? resolved[0];
+ }
+ catch (Exception e) when (!(e is TransportException))
+ {
+ throw new TransportException($"Unable to resolve host '{host}'.", e);
+ }
+ }
+
+ private static string GetString(IReadOnlyDictionary<string, string> p, string key, string fallback)
+ {
+ if (p.TryGetValue("udp." + key, out var prefixed))
+ {
+ return prefixed;
+ }
+ return p.TryGetValue(key, out var plain) ? plain : fallback;
+ }
+
+ private static int GetInt(IReadOnlyDictionary<string, string> p, string key, int fallback)
+ {
+ string? raw = p.TryGetValue("udp." + key, out var prefixed)
+ ? prefixed
+ : p.TryGetValue(key, out var plain) ? plain : null;
+ if (raw == null)
+ {
+ return fallback;
+ }
+ if (!int.TryParse(raw, NumberStyles.Integer, CultureInfo.InvariantCulture, out var value))
+ {
+ throw new TransportException($"'{raw}' is not a valid integer for {key}.");
+ }
+ return value;
+ }
+
+ }
+}
diff --git a/plc4net/transports/udp/UdpTransportConfiguration.cs b/plc4net/transports/udp/UdpTransportConfiguration.cs
new file mode 100644
index 0000000000..9c5f4306ba
--- /dev/null
+++ b/plc4net/transports/udp/UdpTransportConfiguration.cs
@@ -0,0 +1,50 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+using org.apache.plc4net.spi.transports;
+
+namespace org.apache.plc4net.transports.udp
+{
+ /// <summary>
+ /// Settings for the UDP transport: default port, local bind address / port, and
+ /// receive buffer capacity.
+ /// </summary>
+ public class UdpTransportConfiguration : ITransportConfiguration
+ {
+ /// <summary>Port used when the connection string does not name one.</summary>
+ public int DefaultPort { get; set; } = -1;
+
+ /// <summary>
+ /// Local address to bind the socket to. Empty binds to the wildcard address for
+ /// the remote address family and lets the OS choose the outbound interface.
+ /// </summary>
+ public string LocalAddress { get; set; } = string.Empty;
+
+ /// <summary>Local port to bind to; 0 lets the OS choose an ephemeral port.</summary>
+ public int LocalPort { get; set; }
+
+ /// <summary>
+ /// Capacity of the user-space ring buffer between the receive loop and the
+ /// codec, not the kernel socket buffer. It is clamped to at least 65535 (one
+ /// maximum-size datagram) by <see cref="UdpTransport.CreateConfiguration"/>.
+ /// </summary>
+ public int ReceiveBufferSize { get; set; } = 81920;
+
+ }
+}
diff --git a/plc4net/transports/udp/UdpTransportInstance.cs b/plc4net/transports/udp/UdpTransportInstance.cs
new file mode 100644
index 0000000000..e63e705580
--- /dev/null
+++ b/plc4net/transports/udp/UdpTransportInstance.cs
@@ -0,0 +1,412 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+using System;
+using System.IO;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+using org.apache.plc4net.spi.generation;
+using org.apache.plc4net.spi.transports;
+
+namespace org.apache.plc4net.transports.udp
+{
+ /// <summary>
+ /// UDP transport: a bound socket, connected to one remote endpoint, with a
+ /// background receive loop feeding a ring buffer.
+ /// </summary>
+ /// <remarks>
+ /// Each <see cref="Write"/> sends one datagram; each datagram received is appended
+ /// to the buffer whole, in arrival order. There is no stream, so a codec that
+ /// wants message boundaries relies on the protocol's own length field
+ /// (KNXnet/IP's total-length header) — the same as over TCP.
+ /// </remarks>
+ public class UdpTransportInstance : BaseTransportInstance, IAsyncTransportInstance
+ {
+ private readonly Socket _socket;
+ private readonly RingBuffer _ringBuffer;
+ private readonly object _readLock = new object();
+ private readonly object _writeLock = new object();
+ private readonly CancellationTokenSource _shutdown = new CancellationTokenSource();
+ private readonly Task _receiveLoop;
+
+ private const int MaxConsecutiveTransientErrors = 20;
+ private const int TransientBackoffMs = 50;
+
+ private int _open = 1;
+ private int _released;
+
+ private volatile Action? _dataListener;
+ private volatile Action<Exception?>? _disconnectListener;
+ private int _listenerThreadId;
+
+ public UdpTransportInstance(IPEndPoint remoteAddress, UdpTransportConfiguration configuration)
+ : base(configuration)
+ {
+ if (remoteAddress == null)
+ {
+ throw new ArgumentNullException(nameof(remoteAddress));
+ }
+
+ _ringBuffer = new RingBuffer(configuration.ReceiveBufferSize);
+
+ Socket? socket = null;
+ try
+ {
+ socket = new Socket(remoteAddress.AddressFamily, SocketType.Dgram, ProtocolType.Udp);
+ var bindAddress = string.IsNullOrEmpty(configuration.LocalAddress)
+ ? (remoteAddress.AddressFamily == AddressFamily.InterNetworkV6
+ ? IPAddress.IPv6Any
+ : IPAddress.Any)
+ : IPAddress.Parse(configuration.LocalAddress);
+ socket.Bind(new IPEndPoint(bindAddress, configuration.LocalPort));
+
+ // A connected datagram socket fixes the Send destination and filters
+ // incoming datagrams to the configured peer.
+ socket.Connect(remoteAddress);
+
+ _socket = socket;
+ RemoteAddress = remoteAddress;
+ LocalAddress = socket.LocalEndPoint as IPEndPoint
+ ?? throw new TransportException("UDP socket has no local endpoint after bind.");
+ }
+ catch (Exception e)
+ {
+ socket?.Dispose();
+ throw new TransportException(
+ $"Failed to open a UDP socket to {remoteAddress.Address}:{remoteAddress.Port} - {e.Message}", e);
+ }
+
+ _receiveLoop = Task.Run(RunReceiveLoopAsync);
+ }
+
+ public IPEndPoint RemoteAddress { get; }
+
+ /// <summary>The local endpoint selected by the bind operation.</summary>
+ public IPEndPoint LocalAddress { get; }
+
+ public override bool IsOpen => Volatile.Read(ref _open) == 1;
+
+ internal bool ResourcesReleased => Volatile.Read(ref _released) == 1;
+
+ /// <summary>Optional diagnostic output: written and read bytes as annotated hex.</summary>
+ public TextWriter? DiagnosticOutput { get; set; }
+
+ public override int GetNumBytesAvailable()
+ {
+ lock (_readLock)
+ {
+ return _ringBuffer.AvailableForReading;
+ }
+ }
+
+ public override byte[] PeekReadableBytes(int numBytes)
+ {
+ if (numBytes <= 0)
+ {
+ return Array.Empty<byte>();
+ }
+ lock (_readLock)
+ {
+ if (_ringBuffer.AvailableForReading < numBytes)
+ {
+ throw new TransportException(
+ $"Requested {numBytes} bytes but only {_ringBuffer.AvailableForReading} available");
+ }
+ return _ringBuffer.Peek(numBytes);
+ }
+ }
+
+ public override byte[] Read(int numBytes)
+ {
+ if (numBytes <= 0)
+ {
+ return Array.Empty<byte>();
+ }
+ byte[] result;
+ lock (_readLock)
+ {
+ if (_ringBuffer.AvailableForReading < numBytes)
+ {
+ throw new TransportException(
+ $"Requested {numBytes} bytes but only {_ringBuffer.AvailableForReading} available");
+ }
+ result = _ringBuffer.Read(numBytes);
+ }
+ WriteHex("UDP RECV", result);
+ return result;
+ }
+
+ public override void Write(byte[] bytes)
+ {
+ if (bytes == null || bytes.Length == 0)
+ {
+ return;
+ }
+ lock (_writeLock)
+ {
+ EnsureOpen();
+ try
+ {
+ // One datagram per call. A short Send on a datagram socket means the
+ // datagram did not fit; treat it as an error rather than fragmenting.
+ var sent = _socket.Send(bytes, 0, bytes.Length, SocketFlags.None);
+ if (sent != bytes.Length)
+ {
+ throw new TransportException(
+ $"UDP datagram truncated: sent {sent} of {bytes.Length} bytes.");
+ }
+ WriteHex("UDP SEND", bytes);
+ }
+ catch (ObjectDisposedException) when (Volatile.Read(ref _open) == 0)
+ {
+ // A concurrent Close() disposed the socket mid-write: normal shutdown.
+ }
+ catch (SocketException e)
+ {
+ throw new TransportException("Failed to send datagram", e);
+ }
+ }
+ }
+
+ public override void Close()
+ {
+ Volatile.Write(ref _open, 0);
+ ReleaseResources(fromReceiveLoop: false);
+ }
+
+ private void ReleaseResources(bool fromReceiveLoop)
+ {
+ if (Interlocked.CompareExchange(ref _released, 1, 0) != 0)
+ {
+ return;
+ }
+
+ _shutdown.Cancel();
+ try
+ {
+ _socket.Dispose();
+ }
+ catch (ObjectDisposedException)
+ {
+ }
+ finally
+ {
+ var insideLoop = fromReceiveLoop
+ || Volatile.Read(ref _listenerThreadId) == Environment.CurrentManagedThreadId;
+ if (!insideLoop && _receiveLoop != null && !_receiveLoop.IsCompleted)
+ {
+ try
+ {
+ _receiveLoop.Wait(TimeSpan.FromSeconds(1));
+ }
+ catch (AggregateException)
+ {
+ // Close is best-effort shutdown; receive-loop errors were already reported.
+ }
+ }
+ _shutdown.Dispose();
+ }
+ }
+
+ public void RegisterDataListener(Action listener)
+ {
+ if (listener == null)
+ {
+ throw new ArgumentNullException(nameof(listener));
+ }
+ _dataListener = listener;
+
+ // A datagram may have arrived and been buffered before the listener was
+ // registered; deliver one wake-up so the codec drains what is waiting.
+ bool hasBufferedData;
+ lock (_readLock)
+ {
+ hasBufferedData = _ringBuffer.AvailableForReading > 0;
+ }
+ if (hasBufferedData)
+ {
+ NotifyData();
+ }
+ }
+
+ public void RemoveDataListener() => _dataListener = null;
+
+ public void RegisterDisconnectListener(Action<Exception?> listener) => _disconnectListener = listener;
+
+ public void RemoveDisconnectListener() => _disconnectListener = null;
+
+ private void NotifyDisconnect(Exception cause)
+ {
+ var listener = _disconnectListener;
+ if (listener == null)
+ {
+ return;
+ }
+ Volatile.Write(ref _listenerThreadId, Environment.CurrentManagedThreadId);
+ try
+ {
+ listener(cause);
+ }
+ catch
+ {
+ // A misbehaving listener must not take down the receive loop.
+ }
+ finally
+ {
+ Volatile.Write(ref _listenerThreadId, 0);
+ }
+ }
+
+ private void NotifyData()
+ {
+ var listener = _dataListener;
+ if (listener == null)
+ {
+ return;
+ }
+ Volatile.Write(ref _listenerThreadId, Environment.CurrentManagedThreadId);
+ try
+ {
+ listener();
+ }
+ catch
+ {
+ // Same reasoning as NotifyDisconnect.
+ }
+ finally
+ {
+ Volatile.Write(ref _listenerThreadId, 0);
+ }
+ }
+
+ private async Task RunReceiveLoopAsync()
+ {
+ var datagram = new byte[65535];
+ var transientErrors = 0;
+ try
+ {
+ while (Volatile.Read(ref _open) == 1)
+ {
+ int received;
+ try
+ {
+ received = await _socket
+ .ReceiveAsync(new ArraySegment<byte>(datagram), SocketFlags.None)
+ .ConfigureAwait(false);
+ transientErrors = 0;
+ }
+ catch (SocketException e) when (IsTransientReceiveError(e.SocketErrorCode))
+ {
+ // ICMP port/host-unreachable from the peer (ConnectionReset on
+ // Windows, ConnectionRefused on Linux), or an oversized datagram
+ // (MessageSize). The gateway may just not be up yet; back off and
+ // keep listening rather than tearing the transport down. Escalate
+ // only if the errors do not stop.
+ if (++transientErrors > MaxConsecutiveTransientErrors)
+ {
+ throw;
+ }
+ await Task.Delay(TransientBackoffMs, _shutdown.Token).ConfigureAwait(false);
+ continue;
+ }
+
+ if (received <= 0)
+ {
+ continue;
+ }
+
+ var accepted = false;
+ lock (_readLock)
+ {
+ if (received <= _ringBuffer.Capacity)
+ {
+ if (_ringBuffer.RemainingForWriting < received)
+ {
+ // The codec has not kept up. Discard everything buffered
+ // rather than drop bytes from the middle of a frame - a
+ // partial frame left behind desyncs the length-prefixed
+ // codec permanently. The next datagram starts clean.
+ _ringBuffer.Clear();
+ }
+ _ringBuffer.Write(datagram, 0, received);
+ accepted = true;
+ }
+ }
+
+ if (accepted)
+ {
+ NotifyData();
+ }
+ // else: a datagram larger than the whole buffer can never be
+ // delivered intact; drop it and stay alive.
+ }
+ }
+ catch (OperationCanceledException)
+ {
+ // Close() cancelled us.
+ }
+ catch (ObjectDisposedException)
+ {
+ // Close() disposed the socket while we were awaiting a receive.
+ }
+ catch (Exception e)
+ {
+ if (Volatile.Read(ref _open) == 1)
+ {
+ Volatile.Write(ref _open, 0);
+ NotifyDisconnect(e);
+ }
+ }
+ finally
+ {
+ Volatile.Write(ref _open, 0);
+ ReleaseResources(fromReceiveLoop: true);
+ }
+ }
+
+ private static bool IsTransientReceiveError(SocketError code) =>
+ code == SocketError.ConnectionReset // Windows: ICMP port-unreachable
+ || code == SocketError.ConnectionRefused // Linux: ICMP port-unreachable
+ || code == SocketError.MessageSize // datagram larger than the buffer
+ || code == SocketError.HostUnreachable
+ || code == SocketError.NetworkUnreachable;
+
+ private void WriteHex(string label, byte[] bytes)
+ {
+ var diag = DiagnosticOutput;
+ if (diag == null)
+ {
+ return;
+ }
+ diag.WriteLine($"┌ {label} ({bytes.Length} bytes)");
+ for (var i = 0; i < bytes.Length; i += 16)
+ {
+ diag.Write("│ ");
+ for (var j = i; j < Math.Min(i + 16, bytes.Length); j++)
+ {
+ diag.Write($"{bytes[j]:X2} ");
+ }
+ diag.WriteLine();
+ }
+ diag.WriteLine("└");
+ }
+ }
+}
diff --git a/plc4net/test/spi-test/spi-test.csproj b/plc4net/transports/udp/udp.csproj
similarity index 56%
copy from plc4net/test/spi-test/spi-test.csproj
copy to plc4net/transports/udp/udp.csproj
index 469b3f6b81..7e478ca46e 100644
--- a/plc4net/test/spi-test/spi-test.csproj
+++ b/plc4net/transports/udp/udp.csproj
@@ -20,28 +20,18 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
- <AssemblyName>plc4net-spi-test</AssemblyName>
- <Title>PLC4Net: SPI: Test</Title>
- <IsPackable>false</IsPackable>
- <IsTestProject>true</IsTestProject>
+ <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
+ <AssemblyName>plc4net-transports-udp</AssemblyName>
+ <Title>PLC4Net: Transports: UDP</Title>
</PropertyGroup>
<ItemGroup>
- <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.11.1" />
- <PackageReference Include="xunit" Version="2.9.2" />
- <PackageReference Include="xunit.runner.visualstudio" Version="2.8.2">
- <PrivateAssets>all</PrivateAssets>
- <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
- </PackageReference>
+ <ProjectReference Include="..\..\api\api.csproj" />
+ <ProjectReference Include="..\..\spi\spi.csproj" />
</ItemGroup>
<ItemGroup>
- <ProjectReference Include="..\..\api\api.csproj" />
- <ProjectReference Include="..\..\spi\spi.csproj" />
- <ProjectReference Include="..\..\transports\tcp\tcp.csproj" />
- <ProjectReference Include="..\..\transports\test\test.csproj" />
- <ProjectReference Include="..\..\transports\serial\serial.csproj" />
- <ProjectReference Include="..\..\transports\cotp\cotp.csproj" />
+ <InternalsVisibleTo Include="plc4net-spi-test" />
</ItemGroup>
-</Project>
+</Project>
\ No newline at end of file