Using Akka.DotNet.Streams to answer HTTP POST

[tiago-andrade-dvs]

Hello.
First I am going let it be clear that I am new to both Actor based Systems and Dotnet.

I was told to implement using Akka.Dotnet a System that receives a HTTP POST request, gives a response to said request and then uses the data to send 2 other POST requests. But 1 of the final endpoints can't receive as many requests as the one I am trying to implement.
The Grouped() and GroupWithin() functions seem to be promising for this

I thought about using Akka.Streams because I can then create a Source with .TcpStream()
As I understood the way you Handle a Akka.Streams.Dsl.Tcp.IncomingConnection is with a Flow.

Sure, I want 1 input and 1 output, but in the middle of that I would like to be able to use a Broadcast with 3 "Sinks". 1 for each POST request that I want to make and the other one for the POST response.

I also saw that a Flow can be created based on a FlowShape and the shape based on a Graph. But that didn't get me anywere.

The following code DOES NOT work but it is similar to my attempt.

static async Task Main(string[] args)
        {
            // define an incoming request processing logic
            var actorSystem = ActorSystem.Create("AkkaStreams", @"akka.loglevel = DEBUG");
            var materializer = actorSystem.Materializer();

            Source<Akka.Streams.Dsl.Tcp.IncomingConnection, Task<Akka.Streams.Dsl.Tcp.ServerBinding>> connections =
                actorSystem.TcpStream().Bind("127.0.0.1", 8888);

         var post_1 = Sink.ForEach<ByteString>(/* send first POST */);       // Use Something similar to .Grouped()
         var post_2 = Sink.ForEach<ByteString>(/* send second POST */);  // Use Something similar to .Grouped()

        var response = Flow.Create<ByteString>(/* send HTTP response */);

         var http_reciever = Flow.Create<ByteString>()
                    .Select(c =>
                    {
                        String str_value = Encoding.UTF8.GetString(Encoding.Default.GetBytes(c.ToString()));

                        // str_value.Replace(Environment.NewLine, "\n");
                        string[] body_helper = str_value.Split(Environment.NewLine + Environment.NewLine);

                        return ByteString.FromString((body_helper.Length > 1) ? body_helper[1] : "");
                    });

       await connections.RunForeach(connection =>
                  {
                      Console.WriteLine($"New connection from: {connection.RemoteAddress}");

                Flow<ByteString, ByteString, Akka.NotUsed> fl_shape = Flow.FromGraph(
                    GraphDsl.Create(b =>
                    {                        
                        var broadcast = b.Add(new Broadcast<ByteString>(3));

                        var flow1 = b.From(http_reciever).Via(broadcast).Via(response);
                        var flow2 = b.From(broadcast).Via(post_1);
                        var flow3 = b.From(broadcast).Via(post_2);

                        var final_shape = new FlowShape<ByteString, ByteString>(http_reciever, response); // This. Doesn't Compile. I tried something different but this is what I'd like to do

                        return final_shape; // When this is return would be to run I get a Run Time Error
                    })); 

                connection.HandleWith(final_shape, materializer);
            }, materializer);

Problems I am having:

• There's some major aspect that I am doing wrong but I cannot figure out what that Is (probabbly). A assume using akka.streams.dsl is correct and some of the graph component, right ?
• When testing the .Grouped(int n = 3) function with postman things just got stuck. When I canceled the request on postman then 3 Flow "unstucked" themselves.
• How do I send the HTTP response as fast as possible, even if the POST1 and POST2 respective Sinks (or Flows) are still grouping Stuff

Thx for reading
Hope I can get in the right direction

[Aaronontheweb]

@tiago-andrade-dvs some ideas

1. You need to add your http_reciever stage to the GraphDsl in order to use it - see https://github.com/petabridge/Cluster.WebCrawler/blob/fbe19b11a2726189baf63678bff20fc86795661a/src/WebCrawler.Shared.IO/DownloadCoordinator.cs#L64-L91 - the builder.add calls are key.
2. What is the exact stack trace of the exception you're seeing?
3. The await on connections.RunForeach will not complete until the server shuts down - it's designed to block the process from exiting.

I have a sample you might also be interested in looking at: https://github.com/Aaronontheweb/Akka.Streams.Benchmark/blob/dev/src/Akka.Streams.Benchmark/Program.cs

[tiago-andrade-dvs]

Hello @Aaronontheweb.
I am going to answer your sugestions in the order 4, 3, 1, 2:
4. Yes, I saw https://github.com/Aaronontheweb/Akka.Streams.Benchmark/blob/dev/src/Akka.Streams.Benchmark/Program.cs . It was even mentioned on a youtube video that I have seen. The thing is that example has some differences to my problem, thus I wasn't sure if the implementation of what I want is going to end up as something WAY different.
3. Well I think I am using await correctly, right ? I want the http server to run "forever".

1. I didn't know that the context in which you define the stages could impact this much the FlowShape. This type of error seems to be where I am failing. I am also unsure if the FlowShape has to start and end with 1 Source and 1 Sink exactly or if it makes sense to work only with Flow's.
2. Here's the snippet of what I tried after reading suggestion 1

static async Task Main(string[] args)
        {
            // define an incoming request processing logic
            var actorSystem = ActorSystem.Create("AkkaStreams", @"akka.loglevel = DEBUG");
            var materializer = actorSystem.Materializer();

            Source<Akka.Streams.Dsl.Tcp.IncomingConnection, Task<Akka.Streams.Dsl.Tcp.ServerBinding>> connections =
                actorSystem.TcpStream().Bind("127.0.0.1", 8888);

            await connections.RunForeach(connection =>
            {
                Console.WriteLine($"New connection from: {connection.RemoteAddress}");

                Flow<ByteString, ByteString, Akka.NotUsed> fl_shape = Flow.FromGraph(
                    GraphDsl.Create(b =>
                    {
                        Flow<ByteString, ByteString, Akka.NotUsed> http_reciever = Flow.Create<ByteString>()
                            .Select(c =>
                                {
                                        String str_value = Encoding.UTF8.GetString(Encoding.Default.GetBytes(c.ToString()));
                                        // Console.WriteLine(str_value);

                                        // str_value.Replace(Environment.NewLine, "\n");
                                        string[] body_helper = str_value.Split(Environment.NewLine + Environment.NewLine);
                                        // string body = (body_helper.Length > 1) ? body_helper[1] : "";

                                        // JsonSerializer.Serialize();
                                        return ByteString.FromString((body_helper.Length > 1) ? body_helper[1] : "");
                                });

                        var reciever_added = b.Add(http_reciever);

                        var broadcast = b.Add(new Broadcast<ByteString>(3));

                        var post_1 = Flow.Create<ByteString>()
                            .Select(c =>
                                {
                                    // Dont mind if Flow<ByteString, ByteString, Akka.NotUsed> response is "sent" and this isn't done yet
                                    Thread.Sleep(10000);

                                    Console.WriteLine("Starting POST Request 1");
                                    /* etc ...*/
                                    return c;
                                })
                                .Grouped(3)
                                .Select(x => x.Aggregate(ByteString.Empty, (s, byteString) => s.Concat(byteString)));

                        var post_1_added = b.Add(post_1);

                        var post_2 = Flow.Create<ByteString>()
                            .Select(c =>
                                {
                                    // Dont mind if Flow<ByteString, ByteString, Akka.NotUsed> response is "sent" and this isn't done yet
                                    Thread.Sleep(10000);
                                    
                                    Console.WriteLine("Starting POST Request 2");
                                    /* etc ...*/
                                    return c;
                                })
                                .Grouped(3)
                                .Select(x => x.Aggregate(ByteString.Empty, (s, byteString) => s.Concat(byteString)));

                        var post_2_added = b.Add(post_2);

                        Flow<ByteString, ByteString, Akka.NotUsed> response = Flow.Create<ByteString>()
                            .Select(c =>
                                {
                                    string cont = c.ToString();

                                    StringBuilder sb = new StringBuilder("HTTP/1.1");
                                    sb.Append(" 200 OK\n");
                                    sb.Append("Content-Type: text/plain\n");
                                    sb.Append("Content-Length: " + cont.Length);
                                    sb.Append("\n\n");
                                    sb.Append(cont);

                                    return ByteString.FromString(sb.ToString());
                                });
                        var response_added = b.Add(response);

                        var flow_1 = b.From(http_reciever).Via(broadcast).Via(response);
                        var flow_2 = b.From(broadcast).Via(post_1);
                        var flow_3 = b.From(broadcast).Via(post_2);
                        
                        var final_shape = new FlowShape<ByteString, ByteString>(reciever_added.Inlet, response_added.Outlet);
                        return final_shape;
                    }));

                connection.HandleWith(fl_shape, materializer);
            }, materializer);

            // await actorSystem.WhenTerminated;
        }
    }
}

I am also unsure of:

• When I should use the return value of builder.Add() or not
• What are the correct Inlet and Outlet for the final_shape
• Or if using a FlowShape to handle the connection can achieve what I want to do.

This is Stack trace. Everything after the first 3 lines was ONLY after I sent a HTTP Post with Postman

Thank for reading and for the help already given.

[Aaronontheweb]

Well I think I am using await correctly, right ? I want the http server to run "forever".

Ok, cool - just wanted to make sure that was intentional.

[Arkatufus]

@tiago-andrade-dvs

var actorSystem = ActorSystem.Create("AkkaStreams", @"akka.loglevel = DEBUG");
var materializer = actorSystem.Materializer();

Source<Tcp.IncomingConnection, Task<Tcp.ServerBinding>> connections = actorSystem.TcpStream().Bind("127.0.0.1", 8888);

var (serverBind, source) = connections.PreMaterialize(materializer);

var sourceTask = source.RunForeach(connection =>
{
    Console.WriteLine($"New connection from: {connection.RemoteAddress}");

    Flow<ByteString, ByteString, Akka.NotUsed> flShape = Flow.FromGraph(
        GraphDsl.Create(b =>
        {
            var httpReceiver = Flow.Create<ByteString>()
                .Select(c =>
                {
                    var strValue = c.ToString();
                    var bodyHelper = strValue.Split(Environment.NewLine + Environment.NewLine);
                    var result = (bodyHelper.Length > 1) ? bodyHelper[1] : "";
                    result = string.IsNullOrEmpty(result) ? "empty" : result;
                    Console.WriteLine($"Received: {result}");
                    return ByteString.FromString(result);
                });
                
            var post1 = Flow.Create<ByteString>()
                .SelectAsync(1, async c =>
                {
                    // Dont mind if Flow<ByteString, ByteString, Akka.NotUsed> response is "sent" and this isn't done yet
                    Console.WriteLine("Starting POST Request 1");
                    await Task.Delay(1000);
                    /* etc ...*/
                    Console.WriteLine("POST Request 1 Done");

                    return c;
                })
                .Grouped(3)
                .Select(x => x.Aggregate(ByteString.Empty, (s, byteString) => s.Concat(byteString)));

            var post2 = Flow.Create<ByteString>()
                .SelectAsync(1, async c =>
                {
                    // Dont mind if Flow<ByteString, ByteString, Akka.NotUsed> response is "sent" and this isn't done yet
                    Console.WriteLine("Starting POST Request 2");
                    await Task.Delay(1000);
                    /* etc ...*/
                    Console.WriteLine("POST Request 2 Done");

                    return c;
                })
                .Grouped(3)
                .Select(x => x.Aggregate(ByteString.Empty, (s, byteString) => s.Concat(byteString)));

            var response = Flow.Create<ByteString>()
                .Select(c =>
                {
                    var cont = c.ToString();

                    var sb = new StringBuilder("HTTP/1.1");
                    sb.Append(" 200 OK\n");
                    sb.Append("Content-Type: text/plain\n");
                    sb.Append("Content-Length: " + cont.Length);
                    sb.Append("\n\n");
                    sb.Append(cont);

                    Console.WriteLine($"Response: {sb}");
                    return ByteString.FromString(sb.ToString());
                });
            
            var httpReceiverFlow = b.Add(httpReceiver);
            var receiveBroadcast = b.Add(new Broadcast<ByteString>(3));
            var post1Flow = b.Add(post1);
            var post2Flow = b.Add(post2);
            var responseFlow = b.Add(response);

            b.From(httpReceiverFlow).To(receiveBroadcast);
            b.From(receiveBroadcast.Out(0)).To(responseFlow);
            b.From(receiveBroadcast.Out(1))
                .Via(post1Flow)
                .To(Sink.ForEach<ByteString>(bs => Console.WriteLine($"post1: {bs}")));
            b.From(receiveBroadcast.Out(2))
                .Via(post2Flow)
                .To(Sink.ForEach<ByteString>(bs => Console.WriteLine($"post2: {bs}")));

            var finalShape = new FlowShape<ByteString, ByteString>(httpReceiverFlow.Inlet, responseFlow.Outlet);
            return finalShape;
        }));

    connection.HandleWith(flShape, materializer);
}, materializer);

await serverBind;
await actorSystem.WhenTerminated;

[tiago-andrade-dvs]

@Arkatufus
That Solution works beautifully.
Just wondering in what resources did you consult to solve this ? Code examples ? Prior experience with Akka.Dotnet ? Akka.Dotnet docummentation ?
I have read the docs. Not the deepest of dives but I still was unsure on how I would get to this solution.
Thank you BOTH anyway. :)

[Arkatufus]

Honestly, I was only merging the two examples @Aaronontheweb gave above and your requirements. I do have a few years of Akka.NET experience myself, but mainly on the persistence and serialization modules. Stream graph is also a new thing for me.