inabox/InABox.Client.RPC/Transports/Pipe/RPCClientPipeTransport.cs

using System;
using System.Threading;
using InABox.Formatters;
using H.Pipes;
using InABox.Core;

namespace InABox.Rpc
{
    public class RpcClientPipeTransport : RpcClientTransport, IDisposable
    {
        private PipeClient<RpcMessage> _pipe;
        private string _name;
        
        public RpcClientPipeTransport(string name)
        {
            _name = name;
            _pipe = new PipeClient<RpcMessage>(_name, new CoreFormatter<RpcMessage>());
            _pipe.Connected += PipeConnected;
            _pipe.Disconnected += PipeDisconnected;
            _pipe.MessageReceived += PipeMessageReceived;
            _pipe.ExceptionOccurred += PipeExceptionOccurred;
        }

        public override bool Connect(CancellationToken ct = default)
        {
            _pipe.ConnectAsync(ct).Wait(10000, ct);
            return _pipe.IsConnected;
        }

        public override bool IsConnected() => _pipe?.IsConnected == true;
        public override bool IsSecure() => false;
        public override string? ServerName() => _name;
        
        public override void Disconnect()
        {
            _pipe.DisconnectAsync();
        }

        public override void Send(RpcMessage message)
        {
            _pipe.WriteAsync(message).ContinueWith(task =>
            {
                if(task.Exception != null)
                {
                    PipeExceptionOccurred(this, new H.Pipes.Args.ExceptionEventArgs(task.Exception));
                    Fail(message, task.Exception);
                }
            });
        }

        private void PipeConnected(object? sender, H.Pipes.Args.ConnectionEventArgs<RpcMessage> e)
        {
            DoOpen();
        }
        
        private void PipeDisconnected(object? sender, H.Pipes.Args.ConnectionEventArgs<RpcMessage> e)
        {
            DoClose(RpcTransportCloseEventType.Closed);
        }
        
        private void PipeExceptionOccurred(object? sender, H.Pipes.Args.ExceptionEventArgs e)
        {
            Logger.Send(LogType.Error, "", $"Exception occured: {e.Exception.Message}");
        }

        
        private void PipeMessageReceived(object? sender, H.Pipes.Args.ConnectionMessageEventArgs<RpcMessage?> e)
        {
            Accept(e.Message);
        }
        
        public void Dispose()
        {
            _pipe.DisposeAsync().AsTask().Wait();
        }

        ~RpcClientPipeTransport()
        {
            Dispose();
        }
        
        protected override RpcClientTransport Clone() => new RpcClientPipeTransport(_name);
    }
}