inabox/InABox.Client.RPC/Transports/Socket/RPCClientSocketTransport.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using InABox.Clients;
using InABox.Core;
using WebSocketSharp;
using ErrorEventArgs = WebSocketSharp.ErrorEventArgs;

namespace InABox.Rpc
{
    public class RpcClientSocketTransport : RpcClientTransport, IDisposable
    {
        private WebSocket? _socket;
        
        private string[] _urls;

        public RpcClientSocketTransport(string[] urls)
        {
            _urls = urls;
        }
        
        private void Socket_OnOpen(object? sender, EventArgs e)
        {
            DoOpen();
        }
        
        private void Socket_OnMessage(object? sender, MessageEventArgs e)
        {
            RpcMessage? message = null;
            if (e.IsBinary && (e.RawData != null))
                message = Serialization.ReadBinary<RpcMessage>(e.RawData, BinarySerializationSettings.Latest);
            else if (e.IsText && !string.IsNullOrWhiteSpace(e.Data))
                message = Serialization.Deserialize<RpcMessage>(e.Data);
     
            Accept(message);
        }

        private void Socket_OnClose(object? sender, CloseEventArgs e)
        {
            DoClose(RpcTransportCloseEventType.Closed);
        }

        private void Socket_OnError(object? sender, ErrorEventArgs e)
        {
            DoException(e.Exception);
        }
        
        private WebSocket? CreateSocket(string url, bool secure)
        {
            var address = $"{(secure ? "wss" : "ws")}://{url}";
            var socket = new WebSocket(address);            
            
            // Time to wait before disconnect - the default meant that the client disconnected during debugging, since the ping would fail
            socket.WaitTime = TimeSpan.FromSeconds(60);
            socket.OnOpen -= Socket_OnOpen;
            socket.OnError -= Socket_OnError;
            socket.OnClose -= Socket_OnClose;
            socket.OnMessage -= Socket_OnMessage;
            socket.Connect();
            if (socket.ReadyState == WebSocketState.Open)
            {
                DoOpen();

                socket.OnOpen += Socket_OnOpen;
                socket.OnError += Socket_OnError;
                socket.OnClose += Socket_OnClose;
                socket.OnMessage += Socket_OnMessage;

                return socket;
            }
            return null;
        }

        public override void Connect()
        {
            List<Task<WebSocket>> tasks = new List<Task<WebSocket>>();
            foreach (var url in _urls)
            {
                tasks.Add(Task<WebSocket>.Run(() => CreateSocket(url, true)));
                tasks.Add(Task<WebSocket>.Run(() => CreateSocket(url, false)));
            }
            while (tasks.Count > 0)
            {
                var result = Task.WhenAny(tasks).Result;
                if (result.Result == null)
                    tasks.Remove(result);
                else
                {
                    _socket = result.Result;
                    return;
                }
            }
        }

        public override bool IsConnected() => _socket?.ReadyState == WebSocketState.Open;
        public override bool IsSecure() => _socket?.IsSecure == true;
        
        public override void Disconnect()
        {
            _socket?.Close(CloseStatusCode.Normal);
        }

        public override void Send(RpcMessage message)
        {
            var buffer = message.WriteBinary(BinarySerializationSettings.Latest);
            _socket?.Send(buffer);
        }

        protected override RpcClientTransport Clone() => new RpcClientSocketTransport(_urls);

        public void Dispose()
        {
            if (IsConnected())
                Disconnect();
        }
    }
}