tbf/AppDiagnostic/DiagnosticPipeServer.cs

using AppDiagnostic.Logging;
using System;
using System.Collections.Generic;
using System.IO;
using System.IO.Pipes;
using System.Threading;
using System.Threading.Tasks;

namespace AppDiagnostic
{
    public sealed class DiagnosticPipeServer : IDisposable
    {
        private readonly Action<DiagnosticLogEntry> messageReceived;
        private readonly CancellationTokenSource cancellation = new CancellationTokenSource();
        private readonly object syncRoot = new object();
        private readonly List<NamedPipeServerStream> connectedClients = new List<NamedPipeServerStream>();
        private NamedPipeServerStream waitingServer;
        private Task serverTask;

        public DiagnosticPipeServer(Action<DiagnosticLogEntry> messageReceived)
        {
            this.messageReceived = messageReceived ?? throw new ArgumentNullException(nameof(messageReceived));
        }

        public int ConnectedClientCount
        {
            get
            {
                lock (syncRoot)
                {
                    return connectedClients.Count;
                }
            }
        }

        public void Start()
        {
            if (serverTask != null)
            {
                return;
            }

            serverTask = Task.Run((Action)AcceptConnections);
        }

        private void AcceptConnections()
        {
            while (!cancellation.IsCancellationRequested)
            {
                NamedPipeServerStream server = null;

                try
                {
                    server = new NamedPipeServerStream(DiagnosticProtocol.DefaultPipeName, PipeDirection.In, NamedPipeServerStream.MaxAllowedServerInstances, PipeTransmissionMode.Byte, PipeOptions.Asynchronous);

                    lock (syncRoot)
                    {
                        waitingServer = server;
                    }

                    server.WaitForConnection();

                    lock (syncRoot)
                    {
                        waitingServer = null;
                        connectedClients.Add(server);
                    }

                    NamedPipeServerStream connectedServer = server;
                    server = null;
                    Task.Run(() => ReadClient(connectedServer));
                }
                catch (ObjectDisposedException)
                {
                    if (!cancellation.IsCancellationRequested)
                    {
                        throw;
                    }
                }
                catch (IOException)
                {
                    if (!cancellation.IsCancellationRequested)
                    {
                        Thread.Sleep(250);
                    }
                }
                finally
                {
                    lock (syncRoot)
                    {
                        if (ReferenceEquals(waitingServer, server))
                        {
                            waitingServer = null;
                        }
                    }

                    server?.Dispose();
                }
            }
        }

        private void ReadClient(NamedPipeServerStream client)
        {
            try
            {
                while (client.IsConnected && !cancellation.IsCancellationRequested)
                {
                    DiagnosticLogEntry entry = DiagnosticMessageSerializer.ReadFramed(client);
                    messageReceived(entry);
                }
            }
            catch (EndOfStreamException)
            {
                // The producer application was closed.
            }
            catch (IOException)
            {
                // The producer application disconnected.
            }
            catch (ObjectDisposedException)
            {
                // The diagnostic application is shutting down.
            }
            finally
            {
                lock (syncRoot)
                {
                    connectedClients.Remove(client);
                }

                client.Dispose();
            }
        }

        public void Dispose()
        {
            cancellation.Cancel();

            lock (syncRoot)
            {
                waitingServer?.Dispose();
                waitingServer = null;

                foreach (NamedPipeServerStream client in connectedClients.ToArray())
                {
                    client.Dispose();
                }

                connectedClients.Clear();
            }

            try
            {
                serverTask?.Wait(2000);
            }
            catch (AggregateException)
            {
                // Cancellation during shutdown is expected.
            }

            cancellation.Dispose();
        }
    }
}