Skip to content

Commit 3801106

Browse files
drewnoakesdrewnoakes
authored
Merge pull request #1149 from lge-ros2/feature/copilot/claude/opus4_6/implement_UDP
feat: Implement UDP transport support
2 parents 2aa56c7 + 323197d commit 3801106

8 files changed

Lines changed: 1130 additions & 3 deletions

File tree

‎src/NetMQ.Tests/UdpTests.cs‎

Lines changed: 428 additions & 0 deletions
Large diffs are not rendered by default.

‎src/NetMQ/Core/Address.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,12 @@ internal sealed class Address
5858
/// </summary>
5959
public const string EpgmProtocol = "epgm";
6060

61+
/// <summary>
62+
/// The string-literal "udp"
63+
/// - this denotes UDP (User Datagram Protocol) communication.
64+
/// </summary>
65+
public const string UdpProtocol = "udp";
66+
6167
/// <summary>
6268
/// Interface IZAddress specifies that Resolve and property Address must be implemented.
6369
/// </summary>

‎src/NetMQ/Core/Patterns/Radio.cs‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,19 +10,25 @@ internal class Radio : SocketBase
1010
{
1111
private readonly Dictionary<string, HashSet<Pipe>> m_subscriptions;
1212
private readonly Distribution m_distribution;
13+
private readonly HashSet<Pipe> m_udpPipes;
1314

1415
internal Radio(Ctx parent, int threadId, int socketId) : base(parent, threadId, socketId, true)
1516
{
1617
m_options.SocketType = ZmqSocketType.Radio;
1718

1819
m_subscriptions = new Dictionary<string, HashSet<Pipe>>();
1920
m_distribution = new Distribution();
21+
m_udpPipes = new HashSet<Pipe>();
2022
}
2123

2224
protected override void XAttachPipe(Pipe pipe, bool icanhasall)
2325
{
2426
pipe.SetNoDelay();
2527
m_distribution.Attach(pipe);
28+
29+
if (icanhasall)
30+
m_udpPipes.Add(pipe);
31+
2632
XReadActivated(pipe);
2733
}
2834

@@ -67,6 +73,8 @@ protected override void XWriteActivated(Pipe pipe)
6773

6874
protected override void XTerminated(Pipe pipe)
6975
{
76+
m_udpPipes.Remove(pipe);
77+
7078
foreach (var pipes in m_subscriptions.Values)
7179
pipes.Remove(pipe);
7280

@@ -90,6 +98,10 @@ protected override bool XSend(ref Msg msg)
9098
m_distribution.Match(pipe);
9199
}
92100

101+
// For UDP pipes (icanhasall), always match regardless of subscription
102+
foreach (var pipe in m_udpPipes)
103+
m_distribution.Match(pipe);
104+
93105
m_distribution.SendToMatching(ref msg);
94106

95107
return true;

‎src/NetMQ/Core/SessionBase.cs‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ You should have received a copy of the GNU Lesser General Public License
2929
using NetMQ.Core.Transports.Ipc;
3030
using NetMQ.Core.Transports.Pgm;
3131
using NetMQ.Core.Transports.Tcp;
32+
using NetMQ.Core.Transports.Udp;
3233

3334
namespace NetMQ.Core
3435
{
@@ -521,7 +522,8 @@ private void Detached()
521522
// For delayed connect situations, terminate the pipe
522523
// and reestablish later on
523524
if (m_pipe != null && m_options.DelayAttachOnConnect
524-
&& m_addr.Protocol != Address.PgmProtocol && m_addr.Protocol != Address.EpgmProtocol &&
525+
&& m_addr.Protocol != Address.PgmProtocol && m_addr.Protocol != Address.EpgmProtocol
526+
&& m_addr.Protocol != Address.UdpProtocol &&
525527
m_options.SocketType != ZmqSocketType.Peer)
526528
{
527529
m_pipe.Hiccup();
@@ -578,6 +580,13 @@ private void StartConnecting(bool wait)
578580
SendAttach(this, pgmSender);
579581
return;
580582
}
583+
case Address.UdpProtocol:
584+
{
585+
Assumes.NotNull(m_addr.Resolved);
586+
var udpEngine = new UdpEngine(m_options, m_addr, false);
587+
SendAttach(this, udpEngine);
588+
return;
589+
}
581590
}
582591

583592
Debug.Assert(false);

‎src/NetMQ/Core/SocketBase.cs‎

Lines changed: 53 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -290,6 +290,30 @@ private void CheckProtocol(string protocol)
290290
"Multicast protocols are not supported by socket type: " + m_options.SocketType);
291291
}
292292
break;
293+
case Address.UdpProtocol:
294+
// UDP is supported for most socket types.
295+
switch (m_options.SocketType)
296+
{
297+
case ZmqSocketType.Radio:
298+
case ZmqSocketType.Dish:
299+
case ZmqSocketType.Pub:
300+
case ZmqSocketType.Sub:
301+
case ZmqSocketType.Xpub:
302+
case ZmqSocketType.Xsub:
303+
case ZmqSocketType.Req:
304+
case ZmqSocketType.Rep:
305+
case ZmqSocketType.Push:
306+
case ZmqSocketType.Pull:
307+
case ZmqSocketType.Pair:
308+
case ZmqSocketType.Dealer:
309+
case ZmqSocketType.Router:
310+
// All is well
311+
break;
312+
default:
313+
throw new ProtocolNotSupportedException(
314+
"UDP protocol is not supported by socket type: " + m_options.SocketType);
315+
}
316+
break;
293317
default:
294318
throw new ProtocolNotSupportedException("Invalid protocol: " + protocol);
295319
}
@@ -619,6 +643,26 @@ public void Bind(string addr)
619643
throw;
620644
}
621645

646+
m_options.LastEndpoint = listener.Address;
647+
AddEndpoint(addr, listener, null);
648+
break;
649+
}
650+
case Address.UdpProtocol:
651+
{
652+
var listener = new Transports.Udp.UdpListener(ioThread, this, m_options);
653+
654+
try
655+
{
656+
listener.SetAddress(address);
657+
m_port = listener.Port;
658+
}
659+
catch (NetMQException ex)
660+
{
661+
listener.Destroy();
662+
EventBindFailed(addr, ex.ErrorCode);
663+
throw;
664+
}
665+
622666
m_options.LastEndpoint = listener.Address;
623667
AddEndpoint(addr, listener, null);
624668
break;
@@ -832,15 +876,22 @@ public void Connect(string addr)
832876
paddr.Resolved.Resolve(address, m_options.IPv4Only);
833877
break;
834878
}
879+
case Address.UdpProtocol:
880+
{
881+
paddr.Resolved = new Transports.Udp.UdpAddress();
882+
paddr.Resolved.Resolve(address, m_options.IPv4Only);
883+
break;
884+
}
835885
}
836886

837887
// Create session.
838888
SessionBase session = SessionBase.Create(ioThread, true, this, m_options, paddr);
839889
Assumes.NotNull(session);
840890

841-
// PGM does not support subscription forwarding; ask for all data to be
891+
// PGM and UDP do not support subscription forwarding; ask for all data to be
842892
// sent to this pipe.
843-
bool icanhasall = protocol == Address.PgmProtocol || protocol == Address.EpgmProtocol;
893+
bool icanhasall = protocol == Address.PgmProtocol || protocol == Address.EpgmProtocol
894+
|| protocol == Address.UdpProtocol;
844895
Pipe? newPipe = null;
845896

846897
if (!m_options.DelayAttachOnConnect || icanhasall || m_options.SocketType == ZmqSocketType.Peer)

‎src/NetMQ/Core/Transports/Udp/UdpAddress.cs‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
using System;
2+
using System.Linq;
3+
using System.Net;
4+
using System.Net.Sockets;
5+
6+
namespace NetMQ.Core.Transports.Udp
7+
{
8+
internal sealed class UdpAddress : Address.IZAddress
9+
{
10+
public override string ToString()
11+
{
12+
if (Address == null)
13+
return string.Empty;
14+
15+
var endpoint = Address;
16+
17+
return endpoint.AddressFamily == AddressFamily.InterNetworkV6
18+
? Protocol + "://[" + endpoint.Address + "]:" + endpoint.Port
19+
: Protocol + "://" + endpoint.Address + ":" + endpoint.Port;
20+
}
21+
22+
public void Resolve(string name, bool ip4Only)
23+
{
24+
int delimiter = name.LastIndexOf(':');
25+
if (delimiter < 0)
26+
throw new InvalidException($"UdpAddress.Resolve, delimiter ({delimiter}) must be non-negative.");
27+
28+
string addrStr = name.Substring(0, delimiter);
29+
string portStr = name.Substring(delimiter + 1);
30+
31+
if (addrStr.Length >= 2 && addrStr[0] == '[' && addrStr[addrStr.Length - 1] == ']')
32+
addrStr = addrStr.Substring(1, addrStr.Length - 2);
33+
34+
int port;
35+
if (portStr == "*" || portStr == "0")
36+
{
37+
port = 0;
38+
}
39+
else
40+
{
41+
port = Convert.ToInt32(portStr);
42+
if (port == 0)
43+
throw new InvalidException($"UdpAddress.Resolve, port ({portStr}) must be a valid nonzero integer.");
44+
}
45+
46+
IPAddress? ipAddress;
47+
48+
if (addrStr == "*")
49+
{
50+
ipAddress = ip4Only ? IPAddress.Any : IPAddress.IPv6Any;
51+
}
52+
else if (!IPAddress.TryParse(addrStr, out ipAddress))
53+
{
54+
var availableAddresses = Dns.GetHostEntry(addrStr).AddressList;
55+
56+
ipAddress = ip4Only
57+
? availableAddresses.FirstOrDefault(ip => ip.AddressFamily == AddressFamily.InterNetwork)
58+
: availableAddresses.FirstOrDefault(ip =>
59+
ip.AddressFamily == AddressFamily.InterNetwork ||
60+
ip.AddressFamily == AddressFamily.InterNetworkV6);
61+
62+
if (ipAddress == null)
63+
throw new InvalidException($"UdpAddress.Resolve, unable to find an IP address for {name}");
64+
}
65+
66+
Address = new IPEndPoint(ipAddress, port);
67+
}
68+
69+
public IPEndPoint? Address { get; private set; }
70+
71+
public string Protocol => Core.Address.UdpProtocol;
72+
}
73+
}

0 commit comments

Comments
 (0)