Merge pull request #33 from jacqueskang/cancellation-token

support graceful shutdown

Author: jacqueskang

src/IpcServiceSample.ConsoleServer/Program.cs

@@ -2,7 +2,10 @@
 using JKang.IpcServiceFramework;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using System;
 using System.Net;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace IpcServiceSample.ConsoleServer
 {
@@ -14,11 +17,18 @@ static void Main(string[] args)
             IServiceCollection services = ConfigureServices(new ServiceCollection());
 
             // build and run service host
-            new IpcServiceHostBuilder(services.BuildServiceProvider())
+            IIpcServiceHost host = new IpcServiceHostBuilder(services.BuildServiceProvider())
                 .AddNamedPipeEndpoint<IComputingService>("computingEndpoint", "pipeName")
                 .AddTcpEndpoint<ISystemService>("systemEndpoint", IPAddress.Loopback, 45684)
-                .Build()
-                .Run();
+                .Build();
+
+            var source = new CancellationTokenSource();
+            Task.WaitAll(host.RunAsync(source.Token), Task.Run(() =>
+            {
+                Console.WriteLine("Press any key to shutdown.");
+                Console.ReadKey();
+                source.Cancel();
+            }));
         }
 
         private static IServiceCollection ConfigureServices(IServiceCollection services)

src/JKang.IpcServiceFramework.Server/IIpcServiceHost.cs

@@ -1,7 +1,12 @@
-namespace JKang.IpcServiceFramework
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace JKang.IpcServiceFramework
 {
     public interface IIpcServiceHost
     {
         void Run();
+
+        Task RunAsync(CancellationToken cancellationToken = default(CancellationToken));
     }
 }

src/JKang.IpcServiceFramework.Server/IpcServiceEndpoint.cs

@@ -21,7 +21,7 @@ protected IpcServiceEndpoint(string name, IServiceProvider serviceProvider)
         public string Name { get; }
         public IServiceProvider ServiceProvider { get; }
 
-        public abstract void Listen();
+        public abstract Task ListenAsync(CancellationToken cancellationToken = default(CancellationToken));
     }
 
     public abstract class IpcServiceEndpoint<TContract>: IpcServiceEndpoint

src/JKang.IpcServiceFramework.Server/IpcServiceHost.cs

@@ -3,6 +3,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace JKang.IpcServiceFramework
@@ -22,12 +23,15 @@ public IpcServiceHost(IEnumerable<IpcServiceEndpoint> endpoints, IServiceProvide
 
         public void Run()
         {
-            Parallel.ForEach(_endpoints, endpoint =>
-            {
-                _logger?.LogDebug($"Starting endpoint '{endpoint.Name}'...");
-                endpoint.Listen();
-                _logger?.LogDebug($"Endpoint '{endpoint.Name}' stopped.");
-            });
+            RunAsync().Wait();
+        }
+
+        public Task RunAsync(CancellationToken cancellationToken = default(CancellationToken))
+        {
+            Task[] tasks = _endpoints
+                .Select(endpoint => endpoint.ListenAsync(cancellationToken))
+                .ToArray();
+            return Task.WhenAll(tasks);
         }
     }
 }

src/JKang.IpcServiceFramework.Server/NamedPipe/NamedPipeIpcServiceEndpoint.cs

@@ -3,11 +3,12 @@
 using System;
 using System.IO.Pipes;
 using System.Threading;
+using System.Threading.Tasks;
 
 namespace JKang.IpcServiceFramework.NamedPipe
 {
     public class NamedPipeIpcServiceEndpoint<TContract> : IpcServiceEndpoint<TContract>
-        where TContract: class
+        where TContract : class
     {
         private readonly ILogger<NamedPipeIpcServiceEndpoint<TContract>> _logger;
         private readonly NamedPipeOptions _options;
@@ -23,39 +24,51 @@ public NamedPipeIpcServiceEndpoint(string name, IServiceProvider serviceProvider
 
         public string PipeName { get; }
 
-        public override void Listen()
+        public override Task ListenAsync(CancellationToken cancellationToken = default(CancellationToken))
         {
             NamedPipeOptions options = ServiceProvider.GetRequiredService<NamedPipeOptions>();
 
             var threads = new Thread[options.ThreadCount];
             for (int i = 0; i < threads.Length; i++)
             {
                 threads[i] = new Thread(StartServerThread);
-                threads[i].Start();
+                threads[i].Start(cancellationToken);
             }
 
-            while (true)
+            return Task.Factory.StartNew(() =>
             {
-                Thread.Sleep(100);
-                for (int i = 0; i < threads.Length; i++)
+                _logger.LogDebug($"Listening named pipe endpoint '{Name}'...");
+                while (!cancellationToken.IsCancellationRequested)
                 {
-                    if (threads[i].Join(250))
+                    Thread.Sleep(100);
+
+                    for (int i = 0; i < threads.Length; i++)
                     {
-                        // thread is finished, starting a new thread
-                        threads[i] = new Thread(StartServerThread);
-                        threads[i].Start();
+                        if (threads[i].Join(250))
+                        {
+                            // thread is finished, starting a new thread
+                            threads[i] = new Thread(StartServerThread);
+                            threads[i].Start(cancellationToken);
+                        }
                     }
                 }
-            }
+                _logger.LogDebug($"Shutting down named pipe endpoint '{Name}'...");
+            });
         }
 
         private void StartServerThread(object obj)
         {
-            using (var server = new NamedPipeServerStream(PipeName, PipeDirection.InOut, _options.ThreadCount))
+            var token = (CancellationToken)obj;
+            try
+            {
+                using (var server = new NamedPipeServerStream(PipeName, PipeDirection.InOut, _options.ThreadCount))
+                {
+                    server.WaitForConnectionAsync().Wait(token);
+                    Task.Run(() => Process(server, _logger)).Wait(token);
+                }
+            }
+            catch when (token.IsCancellationRequested)
             {
-                server.WaitForConnection();
-
-                Process(server, _logger);
             }
         }
     }

src/JKang.IpcServiceFramework.Server/Tcp/TcpIpcServiceEndpoint.cs

@@ -3,11 +3,13 @@
 using System.Net.Sockets;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.DependencyInjection;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace JKang.IpcServiceFramework.Tcp
 {
     public class TcpIpcServiceEndpoint<TContract> : IpcServiceEndpoint<TContract>
-        where TContract: class
+        where TContract : class
     {
         private readonly ILogger<TcpIpcServiceEndpoint<TContract>> _logger;
         private readonly TcpListener _listener;
@@ -19,18 +21,31 @@ public TcpIpcServiceEndpoint(String name, IServiceProvider serviceProvider, IPAd
             _logger = serviceProvider.GetService<ILogger<TcpIpcServiceEndpoint<TContract>>>();
         }
 
-        public override void Listen()
+        public override Task ListenAsync(CancellationToken cancellationToken = default(CancellationToken))
         {
             _listener.Start();
 
-            _logger?.LogDebug("TCP listener started.");
+            cancellationToken.Register(() =>
+            {
+                _logger.LogDebug($"Shutting down tcp endpoint '{Name}'...");
+                _listener.Stop();
+            });
 
-            while (true)
+            return Task.Run(async () =>
             {
-                TcpClient client = _listener.AcceptTcpClient();
-                NetworkStream server = client.GetStream();
-                Process(server, _logger);
-            }
+                try
+                {
+                    _logger.LogDebug($"Listening tcp endpoint '{Name}'...");
+                    while (true)
+                    {
+                        TcpClient client = await _listener.AcceptTcpClientAsync();
+                        NetworkStream server = client.GetStream();
+                        Process(server, _logger);
+                    }
+                }
+                catch when (cancellationToken.IsCancellationRequested)
+                { }
+            });
         }
     }
 }