inabox/InABox.Client.RPC/Transports/RPCClientTransport.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading;
using InABox.Clients;
using InABox.Core;
using WebSocketSharp;
using Logger = InABox.Core.Logger;

namespace InABox.Rpc
{
    public abstract class RpcClientTransport : IRpcClientTransport
    {
        private RpcClientSession _session = new RpcClientSession();
        private ConcurrentDictionary<Guid, ManualResetEventSlim> _events = new ConcurrentDictionary<Guid, ManualResetEventSlim>();
        private ConcurrentDictionary<Guid, RpcMessage> _responses = new ConcurrentDictionary<Guid, RpcMessage>();

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

        public event RpcTransportOpenEvent? OnOpen;
        protected void DoOpen() => OnOpen?.Invoke(this, new RpcTransportOpenArgs(_session));

        public event RpcTransportCloseEvent? OnClose;
        protected void DoClose(RpcTransportCloseEventType type)
        {
            foreach (var ev in _events)
            {
                _responses.TryAdd(ev.Key, new RpcMessage() { Error = RpcError.DISCONNECTED });
                ev.Value.Set();
            }
            OnClose?.Invoke(this, new RpcTransportCloseArgs(_session, type));
        }
        
        public event RpcTransportExceptionEvent? OnException;
        protected void DoException(Exception e) => OnException?.Invoke(this, new RpcTransportExceptionArgs(_session, e));
        
        public event RpcTransportMessageEvent? OnMessage;
        protected void DoMessage(RpcMessage message) => OnMessage?.Invoke(this, new RpcTransportMessageArgs(_session, message));
        
        public abstract void Connect();
        public abstract bool IsConnected();
        public abstract bool IsSecure();
        
        public abstract void Send(RpcMessage message);

        public void Accept(RpcMessage? message)
        {
            if (message == null)
                return;
            if (_events.TryGetValue(message.Id, out var ev))
            {
                _responses[message.Id] = message;
                ev.Set();
            }
            else
                _responses[message.Id] = message;
        }
        
        public abstract void Disconnect();
        
        public TResult Send<TCommand, TParameters, TResult>(TParameters parameters) 
            where TCommand : IRpcCommand<TParameters,TResult> 
            where TParameters : ISerializeBinary
            where TResult : ISerializeBinary, new()
        {
            var request = new RpcMessage()
            {
                Id = Guid.NewGuid(),
                Command = typeof(TCommand).Name,
                Payload = Serialization.WriteBinary(parameters, BinarySerializationSettings.Latest)
            };
            
            var ev = Queue(request.Id);
            
            Send(request);

            var response = GetResponse(request.Id, ev, DefaultRequestTimeout);
            if (response == null)
                throw new Exception($"{typeof(TCommand).Name}({request.Id}) returned NULL");
            
            if (response.Error != RpcError.NONE)
                throw new Exception($"Exception in {typeof(TCommand).Name}({request.Id}): {response.Error}");
            
            var result = Serialization.ReadBinary<TResult>(response.Payload, BinarySerializationSettings.Latest);
            if (result == null)
                throw new Exception($"Cannot Deserialize {typeof(TCommand).Name}({request.Id})");
            
            return result;
        }
        

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

        public RpcMessage? GetResponse(Guid id, ManualResetEventSlim ev, int timeout)
        {
            if (_responses.TryGetValue(id, out var response))
            {
                _responses.Remove(id, out response);
                _events.Remove(id, out ev);
                return response;
            }

            try
            {
                if (!ev.Wait(timeout))
                {
                    return new RpcMessage() { Error = RpcError.TIMEOUT };
                }
            }
            catch (Exception e)
            {
                Logger.Send(LogType.Error, "", e.Message);
                throw;
            }
            
            _responses.Remove(id, out response);
            _events.Remove(id, out ev);
            return response ?? new RpcMessage() { Error = RpcError.UNKNOWN };
        }

        protected abstract RpcClientTransport Clone();
        
        public bool Ping()
        {
            bool result = false;
            try
            {
                var transport = Clone();
                transport.Connect();
                result = transport.IsConnected();
                transport.Disconnect();
            }
            catch (Exception e)
            {
            }
            return result;
        }

        public DatabaseInfo? Info()
        {
            DatabaseInfo? result = null;
            try
            {
                var transport = Clone();
                transport.Connect();
                result = transport.Send<RpcInfoCommand,RpcInfoParameters,RpcInfoResult>(new RpcInfoParameters()).Info;
                transport.Disconnect();
            }
            catch (Exception e)
            {
                return null;
            }
            return result;
        }

    }
}