add StopAsync()

Author: smdn

src/Smdn.Net.MuninNode/Smdn.Net.MuninNode/NodeBase.LoggerMessages.cs

@@ -21,6 +21,16 @@ partial class NodeBase {
     eventId: default, // TODO
     formatString: "Started munin-node '{HostName}' on '{EndPoint}'."
   );
+  private static readonly Action<ILogger, string, Exception?> LogStoppingNode = LoggerMessage.Define<string>(
+    LogLevel.Debug,
+    eventId: default, // TODO
+    formatString: "Stopping munin-node '{HostName}'."
+  );
+  private static readonly Action<ILogger, string, Exception?> LogStoppedNode = LoggerMessage.Define<string>(
+    LogLevel.Information,
+    eventId: default, // TODO
+    formatString: "Stopped munin-node '{HostName}'."
+  );
   private static readonly Action<ILogger, Exception?> LogStartedAcceptingConnections = LoggerMessage.Define(
     LogLevel.Information,
     eventId: default, // TODO

src/Smdn.Net.MuninNode/Smdn.Net.MuninNode/NodeBase.Obsolete.cs

@@ -112,6 +112,8 @@ public void Start()
       ? createHandlerValueTask.Result
       : createHandlerValueTask.AsTask().GetAwaiter().GetResult();
 
+    sessionCountdownEvent.Reset();
+
     Logger?.LogInformation("started (end point: {EndPoint})", listener.EndPoint);
   }
 }

src/Smdn.Net.MuninNode/Smdn.Net.MuninNode/NodeBase.cs

@@ -78,6 +78,7 @@ public EndPoint EndPoint {
     }
   }
 
+  private CountdownEvent sessionCountdownEvent = new(initialCount: 1);
   private bool disposed;
 
   protected NodeBase(
@@ -135,6 +136,9 @@ protected virtual async ValueTask DisposeAsyncCore()
 
     protocolHandler = null;
 
+    sessionCountdownEvent?.Dispose();
+    sessionCountdownEvent = null!;
+
     disposed = true;
   }
 
@@ -151,6 +155,9 @@ protected virtual void Dispose(bool disposing)
 
     protocolHandler = null;
 
+    sessionCountdownEvent?.Dispose();
+    sessionCountdownEvent = null!;
+
     disposed = true;
   }
 
@@ -240,6 +247,67 @@ async ValueTask StartAsyncCore()
 
       if (Logger is not null)
         LogStartedNode(Logger, HostName, listener.EndPoint, null);
+
+      sessionCountdownEvent.Reset();
+    }
+  }
+
+  /// <summary>
+  /// Stops accepting connections from clients at the <c>Munin-Node</c> currently running.
+  /// </summary>
+  /// <param name="cancellationToken">
+  /// The <see cref="CancellationToken"/> to monitor for cancellation requests.
+  /// </param>
+  /// <returns>
+  /// The <see cref="ValueTask"/> that represents the asynchronous operation,
+  /// starting the <c>Munin-Node</c> instance.
+  /// </returns>
+  /// <remarks>
+  /// If current <c>Munin-Node</c> has already stopped, this method does nothing and returns the result.
+  /// </remarks>
+  public ValueTask StopAsync(CancellationToken cancellationToken)
+  {
+    ThrowIfDisposed();
+
+    if (listener is null)
+      return default; // already stopped
+
+    return StopAsyncCore();
+
+    async ValueTask StopAsyncCore()
+    {
+      if (Logger is not null)
+        LogStoppingNode(Logger, HostName, null);
+
+      // decrement by the initial value of 1 (re)set in Start()/StartAsync()
+      sessionCountdownEvent.Signal();
+
+      try {
+        // wait for all sessions to complete
+        sessionCountdownEvent.Wait(cancellationToken);
+      }
+      catch (OperationCanceledException ex) when (cancellationToken.Equals(ex.CancellationToken)) {
+        // revert decremented counter value
+        sessionCountdownEvent.AddCount();
+
+        throw;
+      }
+
+      await listener.DisposeAsync().ConfigureAwait(false);
+
+      listener = null;
+
+      if (protocolHandler is not null) {
+        if (protocolHandler is IAsyncDisposable asyncDisposableProtocolHandler)
+          await asyncDisposableProtocolHandler.DisposeAsync().ConfigureAwait(false);
+        else if (protocolHandler is IDisposable disposableProtocolHandler)
+          disposableProtocolHandler.Dispose();
+
+        protocolHandler = null;
+      }
+
+      if (Logger is not null)
+        LogStoppedNode(Logger, HostName, null);
     }
   }
 
@@ -335,6 +403,8 @@ public async ValueTask AcceptSingleSessionAsync(
     using var scope = Logger?.BeginScope(client.EndPoint);
 
     try {
+      sessionCountdownEvent.AddCount();
+
       cancellationToken.ThrowIfCancellationRequested();
 
       if (!CanAccept(client))
@@ -348,6 +418,8 @@ await ProcessSessionAsync(
     finally {
       await client.DisposeAsync().ConfigureAwait(false);
 
+      sessionCountdownEvent.Signal();
+
       if (Logger is not null)
         LogAcceptedConnectionClosed(Logger, null);
     }

tests/Smdn.Net.MuninNode/Smdn.Net.MuninNode/NodeBase.cs

@@ -7,6 +7,7 @@
 using System.Net;
 using System.Net.Sockets;
 using System.Runtime.InteropServices;
+using System.Threading;
 using System.Threading.Tasks;
 
 using NUnit.Framework;
@@ -161,6 +162,24 @@ public async Task Start()
 #pragma warning restore CS0618
   }
 
+  [Test]
+  public async Task Start_Restart()
+  {
+    await using var node = CreateNode();
+
+#pragma warning disable CS0618
+    Assert.DoesNotThrow(node.Start);
+    Assert.Throws<InvalidOperationException>(node.Start, "already started");
+#pragma warning restore CS0618
+
+    Assert.That(async () => await node.StopAsync(default), Throws.Nothing);
+
+#pragma warning disable CS0618
+    Assert.DoesNotThrow(node.Start, "restart");
+    Assert.Throws<InvalidOperationException>(node.Start, "already restarted");
+#pragma warning restore CS0618
+  }
+
   [Test]
   public async Task StartAsync()
   {
@@ -179,4 +198,104 @@ public async Task StartAsync()
     Assert.That(() => _ = node.LocalEndPoint, Throws.Nothing, $"{nameof(node.LocalEndPoint)} after start");
 #pragma warning restore CS0618
   }
+
+  [Test]
+  public async Task StartAsync_Restart()
+  {
+    await using var node = CreateNode();
+
+#pragma warning disable CS0618
+    Assert.That(async () => await node.StartAsync(default), Throws.Nothing);
+    Assert.That(async () => await node.StartAsync(default), Throws.InvalidOperationException, "already started");
+#pragma warning restore CS0618
+
+    Assert.That(async () => await node.StopAsync(default), Throws.Nothing);
+
+#pragma warning disable CS0618
+    Assert.That(async () => await node.StartAsync(default), Throws.Nothing, "restart");
+    Assert.That(async () => await node.StartAsync(default), Throws.InvalidOperationException, "already restarted");
+#pragma warning restore CS0618
+  }
+
+  [Test]
+  public async Task StopAsync_StartedByStartAsync()
+  {
+    await using var node = CreateNode();
+
+    Assert.That(async () => await node.StartAsync(default), Throws.Nothing);
+
+    Assert.That(async () => await node.StopAsync(default), Throws.Nothing);
+    Assert.That(async () => await node.StopAsync(default), Throws.Nothing, "stop again");
+  }
+
+  [Test]
+  public async Task StopAsync_StartedByStart()
+  {
+    await using var node = CreateNode();
+
+#pragma warning disable CS0618
+    Assert.That(() => node.Start(), Throws.Nothing);
+#pragma warning restore CS0618
+
+    Assert.That(async () => await node.StopAsync(default), Throws.Nothing);
+    Assert.That(async () => await node.StopAsync(default), Throws.Nothing, "stop again");
+  }
+
+  [Test]
+  public async Task StopAsync_NotStarted()
+  {
+    await using var node = CreateNode();
+
+    Assert.That(async () => await node.StopAsync(default), Throws.Nothing);
+    Assert.That(async () => await node.StopAsync(default), Throws.Nothing, "stop again");
+  }
+
+  [Test]
+  public async Task StopAsync_WhileProcessingClient()
+  {
+    await using var node = CreateNode();
+
+    await node.StartAsync();
+
+    using var cts = new CancellationTokenSource();
+
+    var taskAccept = Task.Run(async () => await node.AcceptAsync(throwIfCancellationRequested: false, cts.Token));
+
+    using var client = CreateClient((IPEndPoint)node.EndPoint, out var writer, out var reader);
+
+    reader.ReadLine(); // banner
+
+    // attempt stop while processing client
+    const int MaxAttempt = 10;
+
+    for (var n = 0; n < MaxAttempt; n++) {
+      using var ctsStopWhileProcessingClientTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(0.1));
+
+      Assert.That(
+        async () => await node.StopAsync(ctsStopWhileProcessingClientTimeout.Token),
+        Throws
+          .InstanceOf<OperationCanceledException>()
+          .With
+          .Property(nameof(OperationCanceledException.CancellationToken))
+          .EqualTo(ctsStopWhileProcessingClientTimeout.Token)
+      );
+    }
+
+    // close session, disconnect client
+    writer.WriteLine(".");
+    writer.Close();
+
+    // stop accepting task
+    cts.Cancel();
+
+    Assert.DoesNotThrowAsync(async () => await taskAccept);
+
+    // attempt stop after processing client
+    using var ctsStopTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(1));
+
+    Assert.That(
+      async () => await node.StopAsync(ctsStopTimeout.Token),
+      Throws.Nothing
+    );
+  }
 }