Migrate HyParView example to AOT

Author: sakno

src/examples/HyParViewPeer/HyParViewClientHandlerFactory.cs

@@ -5,11 +5,11 @@ namespace HyParViewPeer;
 
 internal sealed class HyParViewClientHandlerFactory : IHttpMessageHandlerFactory
 {
-    internal static bool AllowCertificate(object sender, X509Certificate? certificate, X509Chain? chain, SslPolicyErrors sslPolicyErrors) => true;
+    private static bool AllowCertificate(object sender, X509Certificate? certificate, X509Chain? chain, SslPolicyErrors sslPolicyErrors) => true;
 
     public HttpMessageHandler CreateHandler(string name)
     {
-        var handler = new SocketsHttpHandler { ConnectTimeout = TimeSpan.FromMilliseconds(100) };
+        var handler = new SocketsHttpHandler { ConnectTimeout = TimeSpan.FromMilliseconds(1000) };
         handler.SslOptions.RemoteCertificateValidationCallback = AllowCertificate;
         return handler;
     }

src/examples/HyParViewPeer/HyParViewPeer.csproj

@@ -6,7 +6,8 @@
     <LangVersion>latest</LangVersion>
     <Nullable>enable</Nullable>
     <ImplicitUsings>true</ImplicitUsings>
-    <Version>5.11.0</Version>
+    <Version>5.17.0</Version>
+    <PublishAot>true</PublishAot>
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Bench|AnyCPU'">

src/examples/HyParViewPeer/Program.cs

@@ -1,9 +1,17 @@
 using System.Net;
+using System.Net.Http.Headers;
 using System.Reflection;
 using System.Security.Cryptography.X509Certificates;
+using System.Text;
+using DotNext;
+using DotNext.Net;
+using DotNext.Net.Cluster.Discovery.HyParView;
 using DotNext.Net.Cluster.Discovery.HyParView.Http;
-using Microsoft.Extensions.Logging.Console;
-using SslOptions = DotNext.Net.Security.SslOptions;
+using DotNext.Net.Cluster.Messaging.Gossip;
+using DotNext.Net.Http;
+using HyParViewPeer;
+using Microsoft.AspNetCore.Connections;
+using Microsoft.Extensions.Options;
 
 int port;
 int? contactNodePort = null;
@@ -37,19 +45,36 @@
 if (contactNodePort.HasValue)
     configuration.Add("contactNode", $"https://localhost:{contactNodePort.GetValueOrDefault()}");
 
-await new HostBuilder().ConfigureWebHost(webHost =>
+var builder = WebApplication.CreateSlimBuilder();
+builder.Configuration.AddInMemoryCollection(configuration);
+
+// web server
+builder.WebHost.ConfigureKestrel(options =>
 {
-    webHost.UseKestrel(options =>
-    {
-        options.ListenLocalhost(port, static listener => listener.UseHttps(LoadCertificate()));
-    })
-    .UseStartup<HyParViewPeer.Startup>();
-})
-.ConfigureLogging(static builder => builder.AddConsole().SetMinimumLevel(LogLevel.Error))
-.ConfigureAppConfiguration(builder => builder.AddInMemoryCollection(configuration))
-.JoinMesh()
-.Build()
-.RunAsync();
+    options.ListenLocalhost(port, static listener => listener.UseHttps(LoadCertificate()));
+});
+
+// services
+builder.Services
+    .AddSingleton<RumorSpreadingManager>(static sp => new RumorSpreadingManager(EndPointFormatter.UriEndPointComparer))
+    .AddSingleton<IPeerLifetime, HyParViewPeerLifetime>()
+    .AddSingleton<IHttpMessageHandlerFactory, HyParViewClientHandlerFactory>();
+
+// misc
+builder.Logging.AddConsole().SetMinimumLevel(LogLevel.Debug);
+builder.JoinMesh();
+
+await using var app = builder.Build();
+
+// endpoints
+app.UseHyParViewProtocolHandler().UseRouting().UseEndpoints(static endpoints =>
+{
+    endpoints.MapGet(RumorSender.RumorResource, SendRumourAsync);
+    endpoints.MapGet(RumorSender.NeighborsResource, PrintNeighborsAsync);
+    endpoints.MapPost(RumorSender.BroadcastResource, BroadcastAsync);
+});
+
+await app.RunAsync();
 
 static X509Certificate2 LoadCertificate()
 {
@@ -58,4 +83,101 @@ static X509Certificate2 LoadCertificate()
     rawCertificate?.CopyTo(ms);
     ms.Seek(0, SeekOrigin.Begin);
     return new X509Certificate2(ms.ToArray(), "1234");
+}
+
+static (Uri, RumorTimestamp) PrepareMessageId(IServiceProvider sp)
+{
+    var config = sp.GetRequiredService<IOptions<HttpPeerConfiguration>>().Value;
+    var manager = sp.GetRequiredService<RumorSpreadingManager>();
+    return (config.LocalNode!, manager.Tick());
+}
+
+static Task BroadcastAsync(HttpContext context)
+{
+    var senderAddress = RumorSender.ParseSenderAddress(context.Request);
+    var senderId = RumorSender.ParseRumorId(context.Request);
+
+    var spreadingManager = context.RequestServices.GetRequiredService<RumorSpreadingManager>();
+    if (!spreadingManager.CheckOrder(new UriEndPoint(senderAddress), senderId))
+        return Task.CompletedTask;
+
+    Console.WriteLine($"Spreading rumor from {senderAddress} with sequence number = {senderId}");
+
+    return context.RequestServices
+        .GetRequiredService<PeerController>()
+        .EnqueueBroadcastAsync(controller => new RumorSender((IPeerMesh<HttpPeerClient>)controller, senderAddress, senderId))
+        .AsTask();
+}
+
+static Task SendRumourAsync(HttpContext context)
+{
+    var (sender, id) = PrepareMessageId(context.RequestServices);
+
+    return context.RequestServices
+        .GetRequiredService<PeerController>()
+        .EnqueueBroadcastAsync(controller => new RumorSender((IPeerMesh<HttpPeerClient>)controller, sender, id))
+        .AsTask();
+}
+
+static Task PrintNeighborsAsync(HttpContext context)
+{
+    var mesh = context.RequestServices.GetRequiredService<IPeerMesh<HttpPeerClient>>();
+    var sb = new StringBuilder();
+
+    foreach (var peer in mesh.Peers)
+        sb.AppendLine(peer.ToString());
+
+    return context.Response.WriteAsync(sb.ToString(), context.RequestAborted);
+}
+
+file sealed class RumorSender : Disposable, IRumorSender
+{
+    internal const string SenderAddressHeader = "X-Sender-Address";
+    internal const string SenderIdHeader = "X-Rumor-ID";
+
+    internal const string RumorResource = "/rumor";
+    internal const string BroadcastResource = "/broadcast";
+    internal const string NeighborsResource = "/neighbors";
+    
+    private readonly IPeerMesh<HttpPeerClient> mesh;
+    private readonly Uri senderAddress;
+    private readonly RumorTimestamp senderId;
+
+    internal RumorSender(IPeerMesh<HttpPeerClient> mesh, Uri sender, RumorTimestamp id)
+    {
+        this.mesh = mesh;
+        senderAddress = sender;
+        senderId = id;
+    }
+
+    private async Task SendAsync(HttpPeerClient client, CancellationToken token)
+    {
+        using var request = new HttpRequestMessage(HttpMethod.Post, BroadcastResource);
+        AddSenderAddress(request.Headers, senderAddress);
+        AddRumorId(request.Headers, senderId);
+        using var response = await client.SendAsync(request, token);
+        response.EnsureSuccessStatusCode();
+    }
+
+    Task IRumorSender.SendAsync(EndPoint peer, CancellationToken token)
+    {
+        var client = mesh.TryGetPeer(peer);
+        return client is not null && !EndPointFormatter.UriEndPointComparer.Equals(new UriEndPoint(senderAddress), peer)
+            ? SendAsync(client, token)
+            : Task.CompletedTask;
+    }
+
+    public new ValueTask DisposeAsync() => base.DisposeAsync();
+
+    private static void AddSenderAddress(HttpRequestHeaders headers, Uri address)
+        => headers.Add(SenderAddressHeader, address.ToString());
+
+    internal static Uri ParseSenderAddress(HttpRequest request)
+        => new(request.Headers[SenderAddressHeader]!, UriKind.Absolute);
+
+    private static void AddRumorId(HttpRequestHeaders headers, in RumorTimestamp id)
+        => headers.Add(SenderIdHeader, id.ToString());
+
+    internal static RumorTimestamp ParseRumorId(HttpRequest request)
+        => RumorTimestamp.TryParse(request.Headers[SenderIdHeader], out var result) ? result : throw new FormatException("Invalid rumor ID");
 }