C# 实现WebScoket客户端
含登录授权功能
using System;
using System.Text;
using System.Threading;
using System.Collections.Concurrent;
using System.Net.WebSockets;
using Newtonsoft.Json.Linq;
using System.Net;
using System.IO;using MultiCore.ViewModel;namespace MultiCore.Common
{public class WebSocketClient{private static GlobalViewModel globalModel = GlobalViewModel.GetInstance();private static LogHelper log = new LogHelper(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);private static BlockingCollection<string> msgRecords = new BlockingCollection<string>();private static object locker_send = new object();private static bool sendToken = false;private static int seqno;private static string tokenstr = string.Empty;public static void HandleMsgs(){while (true){try{var msg = WebSocketClient.msgRecords.Take();JObject jObj = JObject.Parse(msg);JToken typeToken = jObj["type"];JToken dataToken = jObj["data"];var typestr = typeToken.ToString();var datastr = dataToken.ToString();if (typestr == "Msglog"){globalModel.logRecords.Add(datastr);}else if (typestr == "fi_msg_rsp"){log.LogInfo("Got fi_msg_rsp:" + datastr);JObject jObj2 = JObject.Parse(datastr);JToken codeToken = jObj2["code"];if ((int)codeToken != 0){log.LogError("wrong fi_msg_rsp!");}}else{log.LogError("Unknown websocket message type:" + msg);}}catch (Exception ex){log.LogError("WebSocketClient.HandleMsgs exception:" + ex.Message);log.LogError(ex.StackTrace);}}}public static void Start(){//resetsendToken = false;seqno = 0;tokenstr = string.Empty;var wsCfg = ClientConfig.GetConfig().WebSocket;//用户登录验证，获取tokenif (wsCfg.SkipAuth){tokenstr = "99999999";}else{try{JObject submsg = new JObject();submsg.Add(new JProperty("user_name", wsCfg.Username));submsg.Add(new JProperty("password", wsCfg.Password));JObject sendmsg = new JObject();sendmsg.Add(new JProperty("data", submsg));var senddatastr = sendmsg.ToString();log.LogInfo("http post data:" + senddatastr);var senddata = Encoding.UTF8.GetBytes(senddatastr);string url = string.Format($"http://{wsCfg.Host}:{wsCfg.HTTPPort}/auth/login");var request = (HttpWebRequest)WebRequest.Create(url);request.Method = "POST";request.ContentType = "application/json";var requestStream = request.GetRequestStream();requestStream.Write(senddata, 0, senddata.Length);using (var response = (HttpWebResponse)request.GetResponse()){using (var reader = new StreamReader(response.GetResponseStream())){var content = reader.ReadToEnd();log.LogInfo("http response stream:" + content);JObject jObj = JObject.Parse(content);JToken codeToken = jObj["code"];//JToken msgToken = jObj["msg"];if ((int)codeToken == 0){// cookiestr:"token=7278972820774715393;Path=/, token_5555=7278972820774715393;Path=/"var cookiestr = response.Headers.Get("Set-Cookie");//获取Cookielog.LogInfo("cookiestr:" + cookiestr);int pos1 = cookiestr.IndexOf("token");if (pos1 >= 0){int pos2 = cookiestr.IndexOf("=", pos1);if (pos2 > 0){int pos3 = cookiestr.IndexOf(";", pos2);tokenstr = cookiestr.Substring(pos2 + 1, pos3 - pos2 - 1);}}if (string.IsNullOrWhiteSpace(tokenstr)){log.LogError("tokenstr is null or empty!");LogHelper.LogRecords.Add(new Model.LogInfo(CommonData.GuicmdCode, CommonData.GuiError, "tokenstr is null or empty!"));}}else{log.LogError("http auth error!" + content);LogHelper.LogRecords.Add(new Model.LogInfo(CommonData.GuicmdCode, CommonData.GuiError, "http auth error!" + content));}}}}catch (Exception ex){log.LogError("http auth exception:" + ex.Message);LogHelper.LogRecords.Add(new Model.LogInfo(CommonData.GuicmdCode, CommonData.GuiError, "http auth exception:" + ex.Message));}}log.LogInfo("tokenstr:" + tokenstr);if (string.IsNullOrWhiteSpace(tokenstr)){log.LogError("http auth failed!");LogHelper.LogRecords.Add(new Model.LogInfo(CommonData.GuicmdCode, CommonData.GuiError, "http auth failed!"));return;}using (ClientWebSocket wsclient = new ClientWebSocket()){try{var uri = string.Format($"ws://{wsCfg.Host}:{wsCfg.WSPort}");wsclient.ConnectAsync(new Uri(uri), CancellationToken.None).Wait(5 * 1000); // timeout = 5秒log.LogInfo("WebSocket ConnectAsync State: " + wsclient.State);}catch (Exception ex){log.LogError("Websocket connect exception:" + ex.Message);throw ex;}if (wsclient.State != WebSocketState.Open){log.LogError("Websocket state error: " + wsclient.State);throw (new Exception("Websocket state error"));}log.LogInfo("Websocket connected " + wsclient.State);LogHelper.LogRecords.Add(new Model.LogInfo(CommonData.GuiWebsocketErr, CommonData.Guicmd, "Websocket connected state:" + wsclient.State));if (wsCfg.SubMsgTypes.Contains("Msglog")){Subscribe(wsclient, "Msglog");}try{const int BUFLEN = 1024 * 1024;byte[] buf = new byte[BUFLEN];int offset = 0;while (wsclient.State == WebSocketState.Open){//var result = await ws.ReceiveAsync(new ArraySegment<byte>(buf, offset, BUFLEN - offset), CancellationToken.None);var task = wsclient.ReceiveAsync(new ArraySegment<byte>(buf, offset, BUFLEN - offset), CancellationToken.None);task.Wait();var result = task.Result;if (result.MessageType == WebSocketMessageType.Close){log.LogError("==WebSocketMessageType.Close== WebSocket State: " + wsclient.State);throw (new Exception("WebSocket is closed by server"));}else{//log.LogInfo("==WebSocket== Received Count:[" + result.Count + "] EndOfMessage:[" + result.EndOfMessage + "] data:" + Encoding.ASCII.GetString(buf, offset, result.Count));//should comment out in prodoffset += result.Count;if (result.EndOfMessage){string msg = Encoding.UTF8.GetString(buf, 0, offset);WebSocketClient.msgRecords.Add(msg);//DateTime dt2 = DateTime.Now;//log.LogInfo(typestr + " spend:" + (dt2 - dt1).TotalMilliseconds);offset = 0;}//else//{// log.LogInfo("WebSocket Received Count:[" + result.Count + "] EndOfMessage:[" // + result.EndOfMessage + "] data:" + Encoding.ASCII.GetString(buf, 0, offset));//}}}}catch (Exception ex){log.LogError("WebSocket Recv excpetion: " + ex.Message);log.LogError("WebSocket state: " + wsclient.State);//wsclient.CloseAsync(WebSocketCloseStatus.NormalClosure, null, CancellationToken.None).Wait();throw ex;}}}private static void Subscribe(ClientWebSocket wsclient, string msgname, int interval_ms = 0){lock (locker_send){if (!sendToken){JObject sendmsg = new JObject();sendmsg.Add(new JProperty("type", "token_sub"));JObject submsg = new JObject();submsg.Add(new JProperty("token", tokenstr));submsg.Add(new JProperty("disable_sub_all", 1));sendmsg.Add(new JProperty("data", submsg));string request = sendmsg.ToString();log.LogInfo(string.Format("WebSocket send request {0}", request));//log.LogInfo(string.Format("WebSocket send tokenmsg {0}", submsg.ToString()));byte[] data = Encoding.ASCII.GetBytes($"{request}");wsclient.SendAsync(new ArraySegment<byte>(data), WebSocketMessageType.Text, true, CancellationToken.None).Wait();sendToken = true;seqno++;}{JObject sendmsg = new JObject();sendmsg.Add(new JProperty("seqno", seqno++));sendmsg.Add(new JProperty("type", "table_action"));JObject submsg = new JObject();submsg.Add(new JProperty("msg_name", msgname));submsg.Add(new JProperty("action", "sub"));submsg.Add(new JProperty("interval_ms", interval_ms));submsg.Add(new JProperty("disable_snap", 1));sendmsg.Add(new JProperty("data", submsg));string request = sendmsg.ToString();log.LogInfo(string.Format("WebSocket send request {0}", request));//log.LogInfo(string.Format("WebSocket send tokenmsg {0}", submsg.ToString()));byte[] data = Encoding.ASCII.GetBytes($"{request}");wsclient.SendAsync(new ArraySegment<byte>(data), WebSocketMessageType.Text, true, CancellationToken.None).Wait();}}}}
}