feat(streams): add NegotiatingServerEngine

Author: st0o0

src/Servus.Akka/Transport/Tcp/Listener/TcpListenerStage.cs

@@ -167,10 +167,22 @@ private async Task InitializeConnectionAsync(TcpClient client)
             var localEndPoint = client.Client.LocalEndPoint!;
             var remoteEndPoint = client.Client.RemoteEndPoint!;
 
+            SecurityInfo? security = null;
+            var protocol = TransportProtocol.Tcp;
+
+            if (stream is SslStream sslStream)
+            {
+                security = new SecurityInfo(
+                    sslStream.SslProtocol,
+                    sslStream.NegotiatedApplicationProtocol);
+                protocol = TransportProtocol.Tls;
+            }
+
             var connectionInfo = new ConnectionInfo(
                 localEndPoint,
                 remoteEndPoint,
-                TransportProtocol.Tcp);
+                protocol,
+                security);
 
             var connectionFlow = Flow.FromGraph(
                 new TcpServerConnectionStage(stream, connectionInfo));

src/TurboHTTP.Tests/Streams/ProtocolRouterSpec.cs

@@ -79,4 +79,13 @@ public void ResolveEngine_should_return_http11_for_unknown_version()
         Assert.NotNull(engine);
         Assert.IsType<Http11ServerEngine>(engine);
     }
+
+    [Fact(Timeout = 5000)]
+    public void ResolveNegotiating_should_return_negotiating_engine()
+    {
+        var engine = ProtocolRouter.ResolveNegotiating(DefaultOptions);
+
+        Assert.NotNull(engine);
+        Assert.IsType<NegotiatingServerEngine>(engine);
+    }
 }

src/TurboHTTP/Streams/Lifecycle/ListenerActor.cs

@@ -182,12 +182,7 @@ private IServerProtocolEngine ResolveEngineForListener()
             return ProtocolRouter.ResolveEngine(new Version(3, 0), _serverOptions);
         }
 
-        if (_listenerOptions is TcpListenerOptions { ApplicationProtocols: [var preferred, ..] })
-        {
-            return ProtocolRouter.ResolveEngine(preferred, _serverOptions);
-        }
-
-        return ProtocolRouter.ResolveEngine(new Version(1, 1), _serverOptions);
+        return ProtocolRouter.ResolveNegotiating(_serverOptions);
     }
 
     public static Props Create(

src/TurboHTTP/Streams/NegotiatingServerEngine.cs

@@ -0,0 +1,36 @@
+using Akka;
+using Akka.Streams;
+using Akka.Streams.Dsl;
+using Servus.Akka.Transport;
+using TurboHTTP.Server;
+using TurboHTTP.Streams.Stages.Server;
+
+namespace TurboHTTP.Streams;
+
+internal sealed class NegotiatingServerEngine : IServerProtocolEngine
+{
+    private readonly TurboServerOptions _options;
+
+    public NegotiatingServerEngine(TurboServerOptions options)
+    {
+        _options = options;
+    }
+
+    public BidiFlow<ITransportInbound, HttpRequestMessage, HttpResponseMessage, ITransportOutbound, NotUsed> CreateFlow()
+    {
+        return BidiFlow.FromGraph(GraphDsl.Create(b =>
+        {
+            var connection = b.Add(new ProtocolNegotiatorConnectionStage(_options));
+
+            return new BidiShape<
+                ITransportInbound,
+                HttpRequestMessage,
+                HttpResponseMessage,
+                ITransportOutbound>(
+                connection.InNetwork,
+                connection.OutRequest,
+                connection.InResponse,
+                connection.OutNetwork);
+        }));
+    }
+}

src/TurboHTTP/Streams/ProtocolRouter.cs

@@ -23,4 +23,9 @@ internal static IServerProtocolEngine ResolveEngine(Version version, TurboServer
             _ => new Http11ServerEngine(options)
         };
     }
+
+    internal static IServerProtocolEngine ResolveNegotiating(TurboServerOptions options)
+    {
+        return new NegotiatingServerEngine(options);
+    }
 }

src/TurboHTTP/Streams/Stages/Server/ProtocolNegotiatorConnectionStage.cs

@@ -0,0 +1,27 @@
+using Akka.Streams;
+using Akka.Streams.Stage;
+using Servus.Akka.Transport;
+using TurboHTTP.Protocol;
+using TurboHTTP.Server;
+
+namespace TurboHTTP.Streams.Stages.Server;
+
+internal sealed class ProtocolNegotiatorConnectionStage : GraphStage<ServerConnectionShape>
+{
+    private readonly Inlet<ITransportInbound> _inNetwork = new("NegotiatorConnection.In.Network");
+    private readonly Outlet<HttpRequestMessage> _outRequest = new("NegotiatorConnection.Out.Request");
+    private readonly Inlet<HttpResponseMessage> _inResponse = new("NegotiatorConnection.In.Response");
+    private readonly Outlet<ITransportOutbound> _outNetwork = new("NegotiatorConnection.Out.Network");
+    private readonly TurboServerOptions _options;
+
+    public ProtocolNegotiatorConnectionStage(TurboServerOptions options)
+    {
+        _options = options;
+    }
+
+    public override ServerConnectionShape Shape => new(_inNetwork, _outRequest, _inResponse, _outNetwork);
+
+    protected override GraphStageLogic CreateLogic(Attributes inheritedAttributes)
+        => new HttpConnectionServerStageLogic<ProtocolNegotiatingStateMachine>(this,
+            ops => new ProtocolNegotiatingStateMachine(_options, ops));
+}