Added new projects of new RPC core: MyClient and MtServivce

2026-08-01 21:17:48 +00:00 · 2024-01-04 18:35:15 +02:00
parent 2a13699900
commit feb0192b11
24 changed files with 2078 additions and 0 deletions
@@ -0,0 +1,184 @@
+using System.Net.WebSockets;
+using System.Text;
+
+namespace MtClient
+{
+    public class MtRpcClient
+    {
+        public MtRpcClient(string host, int port)
+        {
+            host_ = host;
+            port_ = port;
+
+            receiveThread_ = new Thread(new ThreadStart(DoReceive));
+            sendThread_ = new Thread(new ThreadStart(DoWrite));
+        }
+
+        public async Task Connect()
+        {
+            Log($"Connect: started to {host_}:{port_}");
+
+            try
+            {
+                await ws_.ConnectAsync(new Uri($"ws://{host_}:{port_}/ws"), CancellationToken.None);
+            }
+            catch (Exception ex)
+            {
+                Log($"Connect failed: {ex.Message}");
+                throw new Exception($"Failed connection to {host_}:{port_}");
+            }
+
+            receiveThread_.Start();
+            sendThread_.Start();
+
+            Log($"Connect: success.");
+        }
+
+        public async void Disconnect()
+        {
+            Log($"Disconnect: {host_}:{port_}");
+
+            try
+            {
+                await ws_.CloseAsync(WebSocketCloseStatus.NormalClosure, null, CancellationToken.None);
+                ws_.Dispose();
+            }
+            catch (Exception ex)
+            {
+                Log($"Disconnect: {ex.Message}");
+            }
+
+            sendWaiter_.Set();
+            sendThread_.Join();
+            receiveThread_.Join();
+
+            Log($"Disconnect: success");
+        }
+
+        public void Send(MtMessage message)
+        {
+            pendingMessages_.Enqueue(message.Serialize());
+            sendWaiter_.Set();
+        }
+
+        private async void DoWrite()
+        {
+            while(ws_.State == WebSocketState.Open)
+            {
+                string? message = null;
+                lock(pendingMessages_)
+                {
+                    if (pendingMessages_.Count > 0)
+                        message = pendingMessages_.Dequeue();
+                }
+
+                if (message == null)
+                {
+                    sendWaiter_.WaitOne();
+                    continue;
+                }                   
+
+                try
+                {
+                    Log($"DoWrite: sending message: {message}");
+                    byte[] bytes = Encoding.ASCII.GetBytes(message);
+                    await ws_.SendAsync(bytes, WebSocketMessageType.Text, true, CancellationToken.None);
+                }
+                catch (Exception e)
+                {
+                    Log($"DoWrite: {e.Message}");
+                }
+            }
+        }
+
+        private async void DoReceive()
+        {
+            try
+            {
+                byte[] recvBuffer = new byte[64 * 1024];
+                while (ws_.State == WebSocketState.Open)
+                {
+                    var result = await ws_.ReceiveAsync(new ArraySegment<byte>(recvBuffer), CancellationToken.None);
+                    if (result.MessageType == WebSocketMessageType.Close)
+                    {
+                        Log($"DoReceive: close signal {result.CloseStatusDescription}");
+                        await ws_.CloseAsync(WebSocketCloseStatus.NormalClosure, null, CancellationToken.None);
+                        break;
+                    }
+                    else
+                    {
+                        var msg = Encoding.ASCII.GetString(recvBuffer, 0, result.Count);
+                        OnReceive(msg);
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                Log($"Exception in receive - {ex.Message}");
+                ConnectionFailed?.Invoke(this, EventArgs.Empty);
+            }
+        }
+
+        private void OnReceive(string msg)
+        {
+            Log($"OnReceive: {msg}");
+
+            if (string.IsNullOrEmpty(msg))
+            {
+                Log("OnReceive: Invalid message (null or empty)");
+                return;
+            }
+
+            var pieces = msg.Split(";", 2);
+
+            if (pieces.Length != 2
+                || string.IsNullOrEmpty(pieces[0])
+                || string.IsNullOrEmpty(pieces[1]))
+            {
+                Log("OnReceive: Invalid message format.");
+                return;
+            }                
+
+            MessageType msgType;
+            try
+            {
+                var msgTypeValue = int.Parse(pieces[0]);
+                msgType = (MessageType)Enum.ToObject(typeof(MessageType), msgTypeValue);
+
+            }
+            catch (Exception e)
+            {
+                Log($"OnReceive: Parse MessageType failed. {e.Message}");
+                return;
+            }
+
+            var message = MtMessageParser.Parse(msgType, (pieces[1]));
+            if (message == null)
+            {
+                Log("OnReceive: Failed parse message payload");
+                return;
+            }
+            
+            MessageReceived?.Invoke(this, message);
+        }
+
+        private void Log(string msg)
+        {
+            Console.WriteLine($"[{Environment.CurrentManagedThreadId}] {msg}");
+        }
+
+        public event EventHandler<MtMessage>? MessageReceived;
+        public event EventHandler<EventArgs>? ConnectionFailed;
+
+        private readonly ClientWebSocket ws_ = new();
+        private readonly string host_;
+        private readonly int port_;
+        private readonly byte[] buf_ = new byte[10000];
+        private readonly Dictionary<MessageType, Func<string, MtMessage?>> msgHandlers_ = new();
+        private readonly Queue<string> pendingMessages_ = [];
+
+        private readonly Thread receiveThread_;
+        private readonly Thread sendThread_;
+        private readonly EventWaitHandle sendWaiter_ = new AutoResetEvent(false);
+    }
+}