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

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using InABox.Clients;
using InABox.Core;
using WebSocket4Net;

namespace InABox.Rpc
{
    public class RpcClientSocketTransport : RpcClientTransport, IDisposable
    {
        private WebSocket? _socket;
        private Task? readTask;
        private ManualResetEventSlim openEvent = new ManualResetEventSlim();

        private string? _host;
        private bool _connected = false;
        
        private string[] _urls;

        public RpcClientSocketTransport(string[] urls)
        {
            _urls = urls;
        }

        // Returns true if we are to continue the receive loop.
        /*private bool DoReceive()
        {
            if(_socket != null)
            {
                try
                {
                    var buffer = new ArraySegment<byte>(new byte[1024]);
                    using (var ms = new MemoryStream())
                    {
                        WebSocketReceiveResult result;
                        do
                        {
                            var task = _socket.ReceiveAsync(buffer, _tokenSource.Token);
                            task.Wait();
                            result = task.Result;

                            ms.Write(buffer.Array, buffer.Offset, result.Count);
                        } while (!result.EndOfMessage);

                        ms.Seek(0, SeekOrigin.Begin);

                        if (result.MessageType == WebSocketMessageType.Close)
                        {
                            if (result.CloseStatus == WebSocketCloseStatus.NormalClosure)
                            {
                                Task.Run(() =>
                                {
                                    DoClose(RpcTransportCloseEventType.Closed);
                                });
                                return false;
                            }
                            else
                            {
                                DoException(new Exception(result.CloseStatusDescription));
                                Task.Run(() =>
                                {
                                    DoClose(RpcTransportCloseEventType.Error);
                                });
                                return false;
                            }
                        }
                        else
                        {
                            RpcMessage? rpcMessage = null;
                            if (result.MessageType == WebSocketMessageType.Binary)
                            {
                                rpcMessage = Serialization.ReadBinary<RpcMessage>(ms, BinarySerializationSettings.Latest);
                            }
                            else if (result.MessageType == WebSocketMessageType.Text)
                            {
                                rpcMessage = Serialization.Deserialize<RpcMessage>(ms);
                            }
                            Accept(rpcMessage);
                        }
                    }
                }
                catch(Exception e)
                {
                    DoException(e);
                    if (!IsConnected())
                    {
                        Task.Run(() =>
                        {
                            DoClose(RpcTransportCloseEventType.Error);
                        });
                    }
                    return false;
                }
            }
            return true;
        }*/
        
        private WebSocket? CreateSocket(string url, bool secure)
        {
            //WebsocketClient client;

            //WebSocket socket = null;
            var address = $"{(secure ? "wss" : "ws")}://{url}";
            var client = new WebSocket(address);

            var openEvent = new ManualResetEventSlim();
            var open = false;
            EventHandler onOpen = (s, e) =>
            {
                open = true;
                openEvent.Set();
            };
            EventHandler onClose = (s, e) =>
            {
                openEvent.Set();
            };

            client.Opened += onOpen;
            client.Closed += onClose;

            client.Error += Client_Error;
            client.MessageReceived += Client_MessageReceived;
            client.DataReceived += Client_DataReceived;
            client.Open();

            openEvent.Wait();

            if (!open)
            {
                return null;
            }

            client.Opened -= onOpen;
            client.Closed -= onClose;

            client.Opened += Client_Opened;
            client.Closed += Client_Closed;

            DoOpen();
            _host = url;

            // Time to wait before disconnect - the default meant that the client disconnected during debugging, since the ping would fail
            /*socket.WaitTime = TimeSpan.FromSeconds(20);
            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 client;
        }

        private void Client_Closed(object sender, EventArgs e)
        {
            openEvent.Set();
            if (_connected)
            {
                _connected = false;
                DoClose(RpcTransportCloseEventType.Closed);
            }
        }

        private void Client_DataReceived(object sender, DataReceivedEventArgs e)
        {
            var rpcMessage = Serialization.ReadBinary<RpcMessage>(e.Data, BinarySerializationSettings.Latest);
            Accept(rpcMessage);
        }

        private void Client_MessageReceived(object sender, MessageReceivedEventArgs e)
        {
            var rpcMessage = Serialization.Deserialize<RpcMessage>(e.Message);
            Accept(rpcMessage);
        }

        private void Client_Error(object sender, SuperSocket.ClientEngine.ErrorEventArgs e)
        {
            DoException(e.Exception);
        }

        private void Client_Opened(object sender, EventArgs e)
        {
            _connected = true;
            DoOpen();
            openEvent.Set();
        }

        public override void Connect()
        {
            if(_socket != null)
            {
                openEvent.Reset();
                _socket.Open();
                openEvent.Wait();
            }
            else
            {
                var tasks = new List<Task<WebSocket?>>();
                foreach (var url in _urls)
                {
                    tasks.Add(Task.Run(() => CreateSocket(url, true)));
                    tasks.Add(Task.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;
                        _connected = true;

                        /*Task.Run(() =>
                        {
                            while (IsConnected())
                            {
                                if (!DoReceive())
                                {
                                    break;
                                }
                            }
                        });*/

                        return;
                    }
                }
            }

        }

        public override bool IsConnected() => _connected;//_socket?.State == WebSocketState.Open;
        public override bool IsSecure() => _socket?.Security.Certificates.Count > 0;

        public override String? ServerName() => _host;
        
        public override void Disconnect()
        {
            _socket?.Close();
            //_socket?.CloseAsync(WebSocketCloseStatus.NormalClosure, "", _tokenSource.Token).Wait();
        }

        public override void Send(RpcMessage message)
        {
            var buffer = message.WriteBinary(BinarySerializationSettings.Latest);
            _socket?.Send(buffer, 0, buffer.Length);//  ?.SendAsync(buffer, WebSocketMessageType.Binary, true, _tokenSource.Token)?.Wait();
        }

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

        public void Dispose()
        {
            _socket?.Close();
            _socket?.Dispose();
            /*if (IsConnected())
                Disconnect();
            _tokenSource.Cancel();
            _socket?.Dispose();*/
        }
    }
}