How listen to connected and other messages on a Tcp connection using TcpStream ?

[ironhide91]

My current code does the following -

1. Creates IO Tcp actor to connect to a remote Tcp endpoint
2. In response to a particular message establishes connection to remote endpoint.
3. I also subscribe to CommandFailed, ConnectionClosed, Connected & Received messages.
4. On Connected message I send a handshake request to remote endpoint.
5. In response to a correct handshake the remote endpoint sends a matching handshake response.
6. Then we follow predefined length based protocol to exchange binary data.

What I want to achieve - Mechanism to process length pre-fixed binary data.

Initial thought was to use System.IO.Pipelines

1. My initial thought was to create pipeline from System.IO.Pipelines and on each Received message write the ByteString data to the pipeline.
2. On the other side of the pipeline I will decode the length prefix and frame individual message and send to downstream for further processing.

I later realised that this can be achieved using Akka.Streams

1. I then switch to Context.System.TcpStream().OutgoingConnection(endPoint.Host, endPoint.Port)
2. But now I'm unable to listen to any of the messages.

using Akka.Actor;
using Akka.IO;
using Akka.Streams.Dsl;
using Akka.Streams;
using System;
using System.Buffers;
using System.Collections.Generic;
using System.Net;
using System.Text;
using IoTcp = Akka.IO.Tcp;

namespace Engine.Actors
{
    public class PeerActor : ReceiveActor
    {
        private readonly DnsEndPoint endPoint;
        private readonly ReadOnlySequence<byte> handShake;

        public static Props Props(DnsEndPoint endPoint, ReadOnlySequence<byte> handShake)
        {
            return Akka.Actor.Props.Create(() => new PeerActor(endPoint, handShake));
        }

        public PeerActor(DnsEndPoint endPoint, ReadOnlySequence<byte> handShake)
        {
            this.endPoint = endPoint;
            this.handShake = handShake;

            Receive<BeginHandShakeCommand>(command => OnBeginHandShakeCommand(command));
            Receive<IoTcp.CommandFailed>(message => OnCommandFailed(message));
            Receive<IoTcp.ConnectionClosed>(message => OnConnectionClosed(message));
            Receive<IoTcp.Connected>(message => OnConnected(message));          
            Receive<IoTcp.Received>(message => OnReceived(message));
        }

        private void OnBeginHandShakeCommand(BeginHandShakeCommand command)
        {
            //Context.System.TcpStream().OutgoingConnection(endPoint.Host, endPoint.Port);
            Context.System.Tcp().Tell(new IoTcp.Connect(endPoint));
        }

        private void OnCommandFailed(IoTcp.CommandFailed message)
        {
            System.Diagnostics.Debug.WriteLine($"command failed on {endPoint}");
        }

        private void OnConnectionClosed(IoTcp.ConnectionClosed message)
        {
            System.Diagnostics.Debug.WriteLine($"connection closed on {endPoint}");
        }

        private void OnConnected(IoTcp.Connected message)
        {
            Sender.Tell(new IoTcp.Register(Self, true, true));

            System.Diagnostics.Debug.WriteLine($"connected to {endPoint}");

            Sender.Tell(
                IoTcp.Write.Create(
                    ByteString.FromBytes(
                        ToArraySegment(handShake))));
        }

        private void OnReceived(IoTcp.Received message)
        {
            System.Diagnostics.Debug.WriteLine($"received on {endPoint}");            
        }

        private IEnumerable<ArraySegment<byte>> ToArraySegment(ReadOnlySequence<byte> ros)
        {
            List<ArraySegment<byte>> arraySegment = new List<ArraySegment<byte>>();

            var temp = ros.GetEnumerator();

            while (temp.MoveNext())
            {
                arraySegment.Add(temp.Current.Span.ToArray());
            }

            return arraySegment;
        }
    }
}

[Horusiath]

If you can use Akka.Streams, you may want to take a look at Framing class. It provides convenience methods for handling incoming chunks of bytes and changing them into frames. Documentation shows how to use it to construct client based on Framing.Delimeter method, but it's not the only one. Framing.SimpleFramingProtocol(maximumMessageLength) is the most generic way of handling what you're after. Short example:

var connection = actorSystem.TcpStream()
    .OutgoingConnection(endpoint)
    .JoinMaterialized(Framing.SimpleFramingProtocol(maximumMessageLength: Int32.MaxValue), Keep.Left);

// 1. handle incoming requests as part of the flow

var handler = connection.SelectAsync(1, async (ByteString request) =>
{
    // handle incoming request
    return request; // respond to a requester - here with echo message
});
var complete = Source.Single(ByteString.FromBytes(handshake.ToArray()))
    .Via(handler)
    .To(Sink.Ignore<ByteString>());
complete.Run(materializer); // execute the connection

// 2. handle incoming connection via queues

var (sender, receiver) = Source.Queue<ByteString>(bufferSize: 4, OverflowStrategy.Backpressure)
    .Via(connection)
    .ToMaterialized(Sink.Queue<ByteString>(), Keep.Both)
    .Run(materializer); // execute connection to obtain access queues
await sender.OfferAsync(ByteString.FromBytes(handshake.ToArray()));
var request = await receiver.PullAsync();
while (request.HasValue)
{
    // handle new request 
    await sender.OfferAsync(response); // respond to the caller
    request = await receiver.PullAsync(); // take the next request
}

Framing.SimpleFramingProtocol basically creates a header around your message in form of [4 bytes length field, Big Endian][User Payload]. If you need length-prefix encoding but with more granular level of control, checkout Framing.LengthField.

[ironhide91]

@Horusiath Thank you for the code snippets. It definitely helps.

How can we handle socket connection and disconnection events/messages within the stream, the way I handle with existing code ?

[Horusiath]

Whenever a connection fails, an exception should be emitted on error lane of the stream. You can read more about error handling in streaming context here: https://getakka.net/articles/streams/error-handling.html

[anpin]

hi, sorry for necroposting, but I can't find a way to obtain Connected and Disconnected messages using streams. I use F# with Akkling , but hopefully my snippet would b e clear

RestartFlow.WithBackoff(
    (fun _ ->
       parent.System
            .TcpStream()
            .OutgoingConnection(
                       remoteAddress = transportState.Endpoint,
                       connectionTimeout = transportState.ConnectionTimeout,
                       idleTimeout = transportState.IdleTimeout)
    ),
    transportState.RestartSettings
    )

  //  this doesn't log connection state, but logs received data 
  // |> Flow.log "[Connection established]" 
 
  |> Flow.viaMat (KillSwitches.Single<ByteString>()) Keep.right
  
  // Recover stage never occurs, due to RestartFlow?  
  // |> Flow.recover (fun exn ->
  //                       logWarningf sys "[Connection failed][%A]" exn
  //                       None) 
  
  |> Flow.joinMat transportState.Flow Keep.left
  |> Graph.run mat

I'm able to successfully recover the stream in case of a failure and keep sending/receiving via transportState.Flow, but neither this stream, nor the transportState.Flow is sending any connection related messages.

Any pointers are highly appreciated.

[Aaronontheweb]

@anpin and that's working well for you?

[anpin]

@Aaronontheweb I guess it depends on how you define well. Haven't run it in production yet and I don't know if it is the right approach or most performant, but in my tests it does the job of exchanging data and recovering the stream in case of failure.