inabox/InABox.Client.RPC/RPCClient.cs

using System.Collections.Concurrent;
using InABox.Clients;
using InABox.Core;

namespace InABox.IPC
{
    
    public class RPCClient<TEntity> : BaseClient<TEntity> where TEntity : Entity, new()
    {
        private IRPCClientTransport _transport;

        private ConcurrentDictionary<Guid, ManualResetEventSlim> Events = new();
        private ConcurrentDictionary<Guid, RPCMessage> 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 RPCClient(Func<IRPCClientTransport> transport)
        {
            _transport = transport();
            _transport.OnOpen += Transport_Opened;
            _transport.OnClose += Transport_Closed;
            _transport.OnException += Transport_Exception;
            _transport.OnMessage += Transport_Message;
            
            _transport.Connect();
        }

        ~RPCClient()
        {
            _transport.OnOpen -= Transport_Opened;
            _transport.OnClose -= Transport_Closed;
            _transport.OnException -= Transport_Exception;
            _transport.OnMessage -= Transport_Message;            
        }
        
        #region TransportManagement
        
        private void Transport_Opened(IRPCTransport transport, RPCTransportOpenArgs e)
        {
            Logger.Send(LogType.Information, "", $"Client Connected");
            Disconnected = false;
            OnConnect?.Invoke();
        }
        
        private void Transport_Message(IRPCTransport transport, RPCTransportMessageArgs e)
        {
            Logger.Send(LogType.Error, "", $"Message received: ({e.Message.Command}) -> {e.Message.Payload}");
        }
        
        private void Transport_Exception(IRPCTransport transport, RPCTransportExceptionArgs e)
        {
            Logger.Send(LogType.Error, "", $"Exception occured: {e.Exception.Message}");
        }

        private void Transport_Closed(IRPCTransport transport, RPCTransportCloseArgs e)
        {
            Logger.Send(LogType.Information, "", $"Client Disconnected");
            Disconnected = true;
            OnDisconnect?.Invoke();
        }
        

        public TResult Send<TCommand, TParameters, TResult>(TParameters properties) where TCommand : IRPCCommand<TParameters,TResult>
        {
            var request = new RPCMessage(
                new Guid(),
                typeof(TCommand).Name,
                Serialization.Serialize(properties)
            );
            var response = Send(request);
            if (response.Error != RPCError.NONE)
                throw new Exception($"Exception in {typeof(TCommand).Name}({request.ID}): {response.Error}");
            var result = Serialization.Deserialize<TResult>(response.Payload);
            if (result == null)
                throw new Exception($"{typeof(TCommand).Name}({request.ID}) returned NULL");
            return result;
        }
        
        public RPCMessage Send(RPCMessage request, int timeout = DefaultRequestTimeout)
        {
            var start = DateTime.Now;
            var ev = Queue(request.ID);
            _transport.Send(request);
            var result = GetResult(request.ID, ev, timeout);
            return result;
        }

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

        public RPCMessage 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
            {
                if (!ev.Wait(timeout))
                {
                    return new RPCMessage(id,"","",RPCError.TIMEOUT);
                }
            }
            catch (Exception e)
            {
                Logger.Send(LogType.Error, "", e.Message);
                throw;
            }
            
            Responses.Remove(id, out result);
            Events.Remove(id, out ev);
            return result ?? new RPCMessage(id,"","",RPCError.UNKNOWN);
        }
        
        #endregion
        
        
        #region Client Interface

        public override DatabaseInfo Info()
        {
            var result = _transport.Send<RPCInfoCommand, RPCInfoParameters, RPCInfoResponse>(new RPCInfoParameters());
            return result.Info;
        }

        private static string[]? _types;
        public override string[] SupportedTypes()
        {
            _types ??= CoreUtils.Entities
                .Where(x => x.GetInterfaces().Contains(typeof(IPersistent)))
                .Select(x => x.EntityName().Replace(".", "_"))
                .ToArray();
            return _types;
        }

        protected override ValidationData DoValidate(string userid, string password, Guid session = default)
        {
            throw new NotImplementedException();
        }

        protected override ValidationData DoValidate(string pin, Guid session = default)
        {
            throw new NotImplementedException();
        }

        protected override ValidationData DoValidate(Guid session = default)
        {
            throw new NotImplementedException();
        }

        protected override CoreTable DoQuery(Filter<TEntity>? filter, Columns<TEntity>? columns, SortOrder<TEntity>? sort = null)
        {
            throw new NotImplementedException();
        }

        protected override TEntity[] DoLoad(Filter<TEntity>? filter = null, SortOrder<TEntity>? sort = null)
        {
            throw new NotImplementedException();
        }

        protected override void DoSave(TEntity entity, string auditnote)
        {
            throw new NotImplementedException();
        }

        protected override void DoSave(IEnumerable<TEntity> entities, string auditnote)
        {
            throw new NotImplementedException();
        }

        protected override void DoDelete(TEntity entity, string auditnote)
        {
            throw new NotImplementedException();
        }

        protected override void DoDelete(IList<TEntity> entities, string auditnote)
        {
            throw new NotImplementedException();
        }

        protected override Dictionary<string, CoreTable> DoQueryMultiple(Dictionary<string, IQueryDef> queries)
        {
            throw new NotImplementedException();
        }

        protected override bool DoCheck2FA(string code, Guid? session)
        {
            throw new NotImplementedException();
        }

        protected override bool DoPing()
        {
            throw new NotImplementedException();
        }
        
        #endregion
    }
}