inabox/inabox.client.ipc/IPCClient.cs

using H.Pipes;
using InABox.Core;
using InABox.IPC.Shared;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace InABox.Client.IPC
{
    internal class IPCClient : IDisposable
    {
        private PipeClient<PipeRequest> Client;

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

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

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

        public bool Disconnected { get; private set; }

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

        public IPCClient(string name)
        {
            Client = new PipeClient<PipeRequest>(name);
            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 PipeRequest Send(PipeRequest request, int timeout = DefaultRequestTimeout)
        {
            var start = DateTime.Now;
            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 PipeRequest GetResult(Guid id, ManualResetEventSlim ev, int timeout)
        {
            if (Responses.TryGetValue(id, out var result))
            {
                Responses.Remove(id, out result);
                Events.Remove(id, out ev);
                return result;
            }

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

        private void Client_MessageReceived(object? sender, H.Pipes.Args.ConnectionMessageEventArgs<PipeRequest?> e)
        {
            if (Events.TryGetValue(e.Message.RequestID, out var ev))
            {
                Responses[e.Message.RequestID] = e.Message;
                ev.Set();
            }
            else
            {
                Responses[e.Message.RequestID] = e.Message;
            }
        }

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

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

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

        ~IPCClient()
        {
            Dispose();
        }
    }
}