inabox/inabox.client.ipc/IPCClientTransport.cs

using H.Pipes;
using InABox.Core;
using InABox.IPC;
using System.Collections.Concurrent;
using InABox.Formatters;

namespace InABox.Client.IPC
{
    public class IPCClientTransport : IDisposable
    {
        private PipeClient<IPCMessage> Client;

        private ConcurrentDictionary<Guid, ManualResetEventSlim> Events = new();
        private ConcurrentDictionary<Guid, IPCMessage> Responses = new();

        private const int DefaultRequestTimeout = 5 * 60 * 1000; // 5 minutes

        public delegate void ConnectEvent();
        public delegate void DisconnectEvent();

        /// <summary>
        /// A handler for any requests pushed from the server, i.e., not initialised by the client.
        /// </summary>
        public delegate void PushEvent(IPCMessage request);

        public bool Disconnected { get; private set; }

        public event ConnectEvent? OnConnect;
        public event DisconnectEvent? OnDisconnect;
        public event PushEvent? OnPush;

        public IPCClientTransport(string pipeName)
        {
            Client = new PipeClient<IPCMessage>(pipeName, new CoreFormatter<IPCMessage>());
            Client.Connected += Client_Connected;
            Client.Disconnected += Client_Disconnected;
            Client.MessageReceived += Client_MessageReceived;
            Client.ExceptionOccurred += Client_ExceptionOccurred;
            Client.ConnectAsync();
        }

        private void Client_ExceptionOccurred(object? sender, H.Pipes.Args.ExceptionEventArgs e)
        {
            Logger.Send(LogType.Error, "", $"Exception occured: {e.Exception.Message}");
        }

        public IPCMessage Send(IPCMessage request, int timeout = DefaultRequestTimeout)
        {
            var ev = Queue(request.RequestID);
            Client.WriteAsync(request);
            var result =  GetResult(request.RequestID, ev, timeout);
            return result;
        }

        public ManualResetEventSlim Queue(Guid id)
        {
            var ev = new ManualResetEventSlim();
            Events[id] = ev;
            return ev;
        }

        public IPCMessage GetResult(Guid id, ManualResetEventSlim ev, int timeout)
        {
            if (Responses.Remove(id, out var result))
            {
                Events.Remove(id, out ev);
                return result;
            }

            try
            {
                if (!ev.Wait(timeout))
                {
                    return IPCMessage.Error(RequestError.TIMEOUT);
                }
            }
            catch (Exception e)
            {
                Console.WriteLine(e);
                throw;
            }
            
            Responses.Remove(id, out result);
            Events.Remove(id, out ev);
            return result ?? IPCMessage.Error(RequestError.UNKNOWN);
        }

        private void Client_MessageReceived(object? sender, H.Pipes.Args.ConnectionMessageEventArgs<IPCMessage?> e)
        {
            if (Events.TryGetValue(e.Message.RequestID, out var ev))
            {
                Responses[e.Message.RequestID] = e.Message;
                ev.Set();
            }
            else
            {
                Task.Run(() =>
                {
                    OnPush?.Invoke(e.Message);
                }).ContinueWith(task =>
                {
                    if (task.Exception != null)
                    {
                        Logger.Send(LogType.Error, "", $"Error in IPC Client Push: {CoreUtils.FormatException(task.Exception)}");
                    }
                });
            }
        }

        private void Client_Connected(object? sender, H.Pipes.Args.ConnectionEventArgs<IPCMessage> e)
        {
            Logger.Send(LogType.Information, "", $"Connected to Pipe: {e.Connection.PipeName}");
            Disconnected = false;
            OnConnect?.Invoke();
        }

        private void Client_Disconnected(object? sender, H.Pipes.Args.ConnectionEventArgs<IPCMessage> e)
        {
            Logger.Send(LogType.Information, "", $"Disconnected from Pipe: {e.Connection.PipeName}");
            foreach (var ev in Events)
            {
                Responses.TryAdd(ev.Key, IPCMessage.Error(RequestError.DISCONNECTED));
                ev.Value.Set();
            }
            Disconnected = true;
            OnDisconnect?.Invoke();
        }

        public void Dispose()
        {
            Client.DisposeAsync().AsTask().Wait();
        }

        ~IPCClientTransport()
        {
            Dispose();
        }
    }
}