UDP listener

[jjmSE]

Hello everyone!

We have a system to collect information from lots of devices (millions) and we use UDPConnected, so a connection is only able to send to the remoteAddress it was connected to, and will receive datagrams only from that address.

We also receive events from these devices so we have a listener, but can't use UDPConnected for that (Bind function is only available for UDP actor). We need to receive all the datagrams in different sockets and received a message when it is closed by the client (similar to TCP Bind). What is the best way to do that?

Thanks!!

[Aaronontheweb]

Hi @jjmSE - let me get back to you on this. Tied up with a long engagement this week but I will reply with some ideas.

[Aaronontheweb]

We also receive events from these devices so we have a listener, but can't use UDPConnected for that (Bind function is only available for UDP actor). We need to receive all the datagrams in different sockets and received a message when it is closed by the client (similar to TCP Bind). What is the best way to do that?

I'm wondering if Akka.Streams.IO.Udp can help with this. I have an example of how to build a duplex TCP client <--> server that I updated for my dotnetconf talk last week: https://github.com/Aaronontheweb/Akka.Streams.Benchmark

Let me check and see how that could be done via UDP in Akka.Streams. This is all built on top of Akka.IO's Udp functionality.

[Aaronontheweb]

Well damn, we do not have a UDP version of this class available:

akka.net/src/core/Akka.Streams/Dsl/Tcp.cs

Lines 24 to 61 in e27220a

	public class Tcp : ExtensionIdProvider<TcpExt>
	{
	/// <summary>
	/// TBD
	/// </summary>
	/// <param name="system">TBD</param>
	/// <returns>TBD</returns>
	public override TcpExt CreateExtension(ExtendedActorSystem system) => new TcpExt(system);
	/// <summary>
	/// Represents a successful TCP server binding.
	/// </summary>
	public struct ServerBinding
	{
	private readonly Func<Task> _unbindAction;
	/// <summary>
	/// Initializes a new instance of the <see cref="ServerBinding"/> class.
	/// </summary>
	/// <param name="localAddress">TBD</param>
	/// <param name="unbindAction">TBD</param>
	public ServerBinding(EndPoint localAddress, Func<Task> unbindAction)
	{
	_unbindAction = unbindAction;
	LocalAddress = localAddress;
	}
	/// <summary>
	/// TBD
	/// </summary>
	public readonly EndPoint LocalAddress;
	/// <summary>
	/// TBD
	/// </summary>
	/// <returns>TBD</returns>
	public Task Unbind() => _unbindAction();
	}

- so it looks like a built-in Akka.Streams solution may be off the table. Let me try something else....

[Aaronontheweb]

akka.net/src/core/Akka.Tests/IO/UdpConnectedIntegrationSpec.cs

Lines 41 to 75 in e27220a

	private IActorRef BindUdp(IPEndPoint address, IActorRef handler)
	{
	var commander = CreateTestProbe();
	commander.Send(Udp.Instance.Apply(Sys).Manager, new Udp.Bind(handler, address));
	commander.ExpectMsg<Udp.Bound>(x => x.LocalAddress.Is(address));
	return commander.Sender;
	}
	private IActorRef ConnectUdp(IPEndPoint localAddress, IPEndPoint remoteAddress, IActorRef handler)
	{
	var commander = CreateTestProbe();
	commander.Send(UdpConnected.Instance.Apply(Sys).Manager, new UdpConnected.Connect(handler, remoteAddress, localAddress));
	commander.ExpectMsg<UdpConnected.Connected>();
	return commander.Sender;
	}
	[Fact]
	public void The_UDP_connection_oriented_implementation_must_be_able_to_send_and_receive_without_binding()
	{
	var serverAddress = _addresses[0];
	var server = BindUdp(serverAddress, TestActor);
	var data1 = ByteString.FromString("To infinity and beyond!");
	var data2 = ByteString.FromString("All your datagram belong to us");
	ConnectUdp(null, serverAddress, TestActor).Tell(UdpConnected.Send.Create(data1));
	var clientAddress = ExpectMsgPf(TimeSpan.FromSeconds(3), "", msg =>
	{
	if (msg is Udp.Received received)
	{
	received.Data.ShouldBe(data1);
	return received.Sender;
	}
	throw new Exception();
	});

So just to make sure I understand why the Udp.Bind command with UdpConnected won't work for you, is it because you don't get notified if the UDP device goes offline? As in, there's no heartbeat detection? It seems like you just use the UDP actor with UdpConnected per the integration tests I linked - can you help me better understand why that won't work?

[ismaelhamed]

I was just looking into this yesterday. The problem is that being Udp a connectionless protocol, there's no way to reliably implement a "disconnected" mechanism. Even the Connect() at the Udp socket level merely sets the default destination address for send, and then tells the OS for which incoming address to actually accept packets. But there's no concept of connection, and obviously no handshake, confirmation, or any way to assess the state of the destination address.

Looks like you typically must build this on top of Udp, like Google's SIP protocol.

[Aaronontheweb]

You could use an Akka.Streams KeepAlive stage to maybe add a heartbeat mechanism over UDP on the server-side. That would work. Otherwise, you'd need a dead man's switch to do it while listening for incoming events from the downstream devices.

[Kiddinglife]

hey @jjmSE looks like an ingress issue. so basically you need write a udp ingress gateway for your devices traffics. I guess even akka.java does not have such built-in gateway. have you tried write a simple aka system that has two interfaces: udp interface for the incoming device data and the other kka.net actor for the outgoing data, with a simple protocol transformation from your custom protocol to akka-native protocol. if you are using k8s, you can use hostport for your ingress docker instance. and then syn the ingress instances to some backend so that your device connects and get the list of healthy ingress instances to do client-side load balance.
Aother similar solution is to directly connec to your actor system that contains your device actor if your actor system can be public-faced to avoid the extra costs to rum those ingress ctor systems.