zhusenlin/serein-flow
1
0
Fork 0
mirror of https://gitee.com/langsisi_admin/serein-flow synced 2026-03-30 21:16:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

修改了很多

Browse Source
This commit is contained in:
fengjiayi
2024-10-27 00:54:10 +08:00
parent 6f26d303e4
commit cb2553ac69
66 changed files with 2215 additions and 1402 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
Library/Network/WebSocket/Handle/JsonMsgHandleConfig.cs
View File

@@ -31,7 +31,7 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
Action<Exception, Action<object>> onExceptionTracking,
bool ArgNotNull)
{
EmitMethodType = EmitHelper.CreateDynamicMethod(methodInfo,out EmitDelegate);
DelegateDetails = new DelegateDetails(methodInfo);
this.Module = model;
Instance = instance;
var parameterInfos = methodInfo.GetParameters();
@@ -72,15 +72,10 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
/// 参数不能为空
/// </summary>
private bool ArgNotNull;
/// <summary>
/// Emit委托
/// </summary>
private readonly Delegate EmitDelegate;
/// <summary>
/// Emit委托类型
/// </summary>
private readonly EmitHelper.EmitMethodType EmitMethodType;
private readonly DelegateDetails DelegateDetails;
/// <summary>
/// 未捕获的异常跟踪
/// </summary>
@@ -114,29 +109,26 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
/// </summary>
private readonly bool[] IsCheckArgNotNull;
//private object ConvertArg(Type type, string argName )
//{
//}
public async void Handle(Func<object, Task> SendAsync,string msgId, JObject jsonObject)
{
object[] args = new object[ParameterType.Length];
bool isCanInvoke = true;; // 表示是否可以调用方法
for (int i = 0; i < ParameterType.Length; i++)
{
var type = ParameterType[i];
var argName = ParameterName[i];
#region DATA JSON数据
if (useData[i])
{
args[i] = jsonObject.ToObject(type);
}
#endregion
else if (useMsgId[i])
var type = ParameterType[i]; // 入参变量类型
var argName = ParameterName[i]; // 入参参数名称
#region 传递消息ID
if (useMsgId[i])
{
args[i] = msgId;
}
#endregion
#region DATA JSON数据
else if (useData[i])
{
args[i] = jsonObject.ToObject(type);
}
#endregion
#region 值类型参数
else if (type.IsValueType)
{
@@ -229,23 +221,7 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
object result;
try
{
if (EmitMethodType == EmitHelper.EmitMethodType.HasResultTask && EmitDelegate is Func<object, object[], Task<object>> hasResultTask)
{
result = await hasResultTask(Instance, args);
}
else if (EmitMethodType == EmitHelper.EmitMethodType.Task && EmitDelegate is Func<object, object[], Task> task)
{
await task.Invoke(Instance, args);
result = null;
}
else if (EmitMethodType == EmitHelper.EmitMethodType.Func && EmitDelegate is Func<object, object[], object> func)
{
result = func.Invoke(Instance, args);
}
else
{
result = null;
}
result = await DelegateDetails.InvokeAsync(Instance, args);
}
catch (Exception ex)
{
@@ -256,25 +232,11 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
await SendAsync.Invoke(exData);
}));
}
//sw.Stop();
//Console.WriteLine($"Emit Invoke：{sw.ElapsedTicks * 1000000F / Stopwatch.Frequency:n3}μs");
if (Module.IsReturnValue)
{
if (result is null)
{
result = "null";
}
_ = SendAsync.Invoke(result);
}
//if( && result != null && result.GetType().IsClass)
//{
// //var reusltJsonText = JsonConvert.SerializeObject(result);
// //_ = SendAsync.Invoke($"{reusltJsonText}");
//}
}

15
Library/Network/WebSocket/Handle/WebSocketHandleModule.cs
View File

@@ -91,6 +91,8 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
MyHandleConfigs.Clear();
}
private HashSet<string> _myMsgIdHash = new HashSet<string>();
/// <summary>
/// 处理JSON数据
/// </summary>
@@ -106,11 +108,15 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
return;
}
string msgId = jsonObject.GetValue(MsgIdJsonKey)?.ToString();
if (_myMsgIdHash.Contains(msgId))
{
Console.WriteLine($"[{msgId}]{theme} 消息重复");
return;
}
_myMsgIdHash.Add(msgId);
try
{
JObject dataObj = jsonObject.GetValue(DataJsonKey)?.ToObject<JObject>();
handldConfig.Handle(async (data) =>
{
@@ -170,9 +176,8 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
}
var msg = jsonData.ToString();
//Console.WriteLine(msg);
//Console.WriteLine();
await sendAsync.Invoke(msg);
}

15
Library/Network/WebSocket/Handle/WebSocketMsgHandleHelper.cs
View File

@@ -155,21 +155,16 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
/// <summary>
/// 异步处理消息
/// </summary>
/// <param name="SendAsync"></param>
/// <param name="sendAsync"></param>
/// <param name="message"></param>
/// <returns></returns>
public async Task HandleMsgAsync(Func<string, Task> SendAsync, string message)
public void HandleMsg(Func<string, Task> sendAsync, string message)
{
//Console.WriteLine(message);
JObject json = JObject.Parse(message);
await Task.Run(() =>
foreach (var module in MyHandleModuleDict.Values)
{
foreach (var module in MyHandleModuleDict.Values)
{
module.HandleSocketMsg(SendAsync, json);
}
});
module.HandleSocketMsg(sendAsync, json);
}
}

24
Library/Network/WebSocket/SocketControlBase.cs
View File

@@ -8,28 +8,4 @@ namespace Serein.Library.Network.WebSocketCommunication
}
//[AutoRegister(RegisterSequence.FlowLoading)]
//[AutoSocketModule(JsonThemeField = "theme", JsonDataField = "data")]
//public class UserService : ISocketControlBase
//{
// public Guid HandleGuid { get; } = new Guid();
// // Action<string> 类型是特殊的，会用一个委托代替，这个委托可以将文本信息发送到客户端
// // Action<object> 类型是特殊的，会用一个委托代替，这个委托可以将对象转成json发送到客户端
// [AutoSocketHandle]
// public void AddUser(User user,Action<string> Recover)
// {
// Console.WriteLine(user.ToString());
// Recover("ok");
// }
// [AutoSocketHandle(ThemeValue = "Remote")]
// public void DeleteUser(User user, Action<string> Recover)
// {
// Console.WriteLine(user.ToString());
// }
//}
}

73
Library/Network/WebSocket/TestExtension.cs Normal file
View File

@@ -0,0 +1,73 @@
using Serein.Library.Utils;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Net.WebSockets;
using System.Text;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
namespace Serein.Library.Network.WebSocketCommunication
{
public class MsgQueueUtil
{
public ConcurrentQueue<string> Msgs = new ConcurrentQueue<string>();
private readonly Channel<string> _msgChannel;
public MsgQueueUtil()
{
_msgChannel = CreateChannel();
}
private Channel<string> CreateChannel()
{
return Channel.CreateBounded<string>(new BoundedChannelOptions(100)
{
FullMode = BoundedChannelFullMode.Wait
});
}
/// <summary>
/// 等待消息
/// </summary>
/// <returns></returns>
public async Task<string> WaitMsgAsync()
{
var state = await _msgChannel.Reader.ReadAsync();
return state;
}
public void WriteMsg(string msg)
{
//Msgs.Enqueue(msg);
Console.WriteLine($"{DateTime.Now}{msg}{Environment.NewLine}");
_ = _msgChannel.Writer.WriteAsync(msg);
}
public bool TryGetMsg(out string msg)
{
return Msgs.TryDequeue(out msg);
}
}
public class SocketExtension
{
/// <summary>
/// 发送消息
/// </summary>
/// <param name="webSocket"></param>
/// <param name="message"></param>
/// <returns></returns>
public static async Task SendAsync(WebSocket webSocket, string message)
{
var buffer = Encoding.UTF8.GetBytes(message);
await webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
}
}
}

164
Library/Network/WebSocket/WebSocketClient.cs
View File

@@ -1,10 +1,13 @@
using Serein.Library.Network.WebSocketCommunication.Handle;
using System;
using System.Diagnostics;
using System.IO.Compression;
using System.IO;
using System.Net.WebSockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using static System.Net.Mime.MediaTypeNames;
namespace Serein.Library.Network.WebSocketCommunication
{
@@ -38,7 +41,6 @@ namespace Serein.Library.Network.WebSocketCommunication
{
try
{
await _client.ConnectAsync(new Uri(uri), CancellationToken.None);
_ = ReceiveAsync();
return true;
@@ -58,8 +60,12 @@ namespace Serein.Library.Network.WebSocketCommunication
/// <returns></returns>
public async Task SendAsync(string message)
{
var buffer = Encoding.UTF8.GetBytes(message);
await _client.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
Console.WriteLine("发送消息");
await Task.Delay(2000);
await SocketExtension.SendAsync(this._client, message); // 回复客户端
Console.WriteLine();
//var buffer = Encoding.UTF8.GetBytes(message);
//await _client.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
}
/// <summary>
@@ -68,13 +74,21 @@ namespace Serein.Library.Network.WebSocketCommunication
/// <returns></returns>
private async Task ReceiveAsync()
{
var buffer = new byte[1024];
var msgQueueUtil = new MsgQueueUtil();
_ = Task.Run(async () =>
{
await HandleMsgAsync(_client, msgQueueUtil);
});
var receivedMessage = new StringBuilder(); // 用于拼接长消息
while (_client.State == WebSocketState.Open)
{
try
{
var buffer = new byte[1024];
WebSocketReceiveResult result;
do
@@ -86,21 +100,22 @@ namespace Serein.Library.Network.WebSocketCommunication
receivedMessage.Append(partialMessage);
} while (!result.EndOfMessage); // 判断是否已经收到完整消息
var message = receivedMessage.ToString();
msgQueueUtil.WriteMsg(message);
receivedMessage.Clear(); // 清空 StringBuilder 为下一条消息做准备
// 处理收到的完整消息
if (result.MessageType == WebSocketMessageType.Close)
{
await _client.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", CancellationToken.None);
}
else
{
var completeMessage = receivedMessage.ToString();
_ = MsgHandleHelper.HandleMsgAsync(SendAsync, completeMessage); // 处理消息
//Debug.WriteLine($"Received: {completeMessage}");
}
// 清空 StringBuilder 为下一条消息做准备
receivedMessage.Clear();
//else
//{
// var completeMessage = receivedMessage.ToString();
// MsgHandleHelper.HandleMsg(SendAsync, completeMessage); // 处理消息，如果方法入参是需要发送消息委托时，将 SendAsync 作为委托参数提供
// //Debug.WriteLine($"Received: {completeMessage}");
//}
}
catch (Exception ex)
{
@@ -110,65 +125,86 @@ namespace Serein.Library.Network.WebSocketCommunication
}
/* #region 消息处理
private readonly string ThemeField;
private readonly ConcurrentDictionary<string, HandldConfig> ThemeConfigs = new ConcurrentDictionary<string, HandldConfig>();
public async Task HandleSocketMsg(string jsonStr)
public async Task HandleMsgAsync(WebSocket webSocket,
MsgQueueUtil msgQueueUtil)
{
JObject json;
try
{
json = JObject.Parse(jsonStr);
}
catch (Exception ex)
{
await SendAsync(_client, ex.Message);
return;
}
// 获取到消息
string themeName = json[ThemeField]?.ToString();
if (!ThemeConfigs.TryGetValue(themeName, out var handldConfig))
{
return;
}
object dataValue;
if (string.IsNullOrEmpty(handldConfig.DataField))
while (true)
{
dataValue = json.ToObject(handldConfig.DataType);
}
else
{
dataValue = json[handldConfig.DataField].ToObject(handldConfig.DataType);
}
await handldConfig.Invoke(dataValue, SendAsync);
}
public void AddConfig(string themeName, Type dataType, MsgHandler msgHandler)
{
if (!ThemeConfigs.TryGetValue(themeName, out var handldConfig))
{
handldConfig = new HandldConfig
var message = await msgQueueUtil.WaitMsgAsync(); // 有消息时通知
//if (!msgQueueUtil.TryGetMsg(out var message)) // 获取消息
//{
// return;
//}
// 消息处理
MsgHandleHelper.HandleMsg(async (text) =>
{
DataField = themeName,
DataType = dataType
};
ThemeConfigs.TryAdd(themeName, handldConfig);
await SocketExtension.SendAsync(webSocket, text); // 回复客户端，处理方法中入参如果需要发送消息委托，则将该回调方法作为委托参数传入
}, message); // 处理消息
}
handldConfig.HandldAsync += msgHandler;
}
public void RemoteConfig(string themeName, MsgHandler msgHandler)
{
if (ThemeConfigs.TryGetValue(themeName, out var handldConfig))
/* #region 消息处理
private readonly string ThemeField;
private readonly ConcurrentDictionary<string, HandldConfig> ThemeConfigs = new ConcurrentDictionary<string, HandldConfig>();
public async Task HandleSocketMsg(string jsonStr)
{
handldConfig.HandldAsync -= msgHandler;
if (!handldConfig.HasSubscribers)
JObject json;
try
{
ThemeConfigs.TryRemove(themeName, out _);
json = JObject.Parse(jsonStr);
}
catch (Exception ex)
{
await SendAsync(_client, ex.Message);
return;
}
// 获取到消息
string themeName = json[ThemeField]?.ToString();
if (!ThemeConfigs.TryGetValue(themeName, out var handldConfig))
{
return;
}
object dataValue;
if (string.IsNullOrEmpty(handldConfig.DataField))
{
dataValue = json.ToObject(handldConfig.DataType);
}
else
{
dataValue = json[handldConfig.DataField].ToObject(handldConfig.DataType);
}
await handldConfig.Invoke(dataValue, SendAsync);
}
public void AddConfig(string themeName, Type dataType, MsgHandler msgHandler)
{
if (!ThemeConfigs.TryGetValue(themeName, out var handldConfig))
{
handldConfig = new HandldConfig
{
DataField = themeName,
DataType = dataType
};
ThemeConfigs.TryAdd(themeName, handldConfig);
}
handldConfig.HandldAsync += msgHandler;
}
public void RemoteConfig(string themeName, MsgHandler msgHandler)
{
if (ThemeConfigs.TryGetValue(themeName, out var handldConfig))
{
handldConfig.HandldAsync -= msgHandler;
if (!handldConfig.HasSubscribers)
{
ThemeConfigs.TryRemove(themeName, out _);
}
}
}
#endregion*/
}
#endregion*/
}
}

109
Library/Network/WebSocket/WebSocketServer.cs
View File

@@ -180,67 +180,46 @@ namespace Serein.Library.Network.WebSocketCommunication
return;
}
var msgQueueUtil = new MsgQueueUtil();
_ = Task.Run(async () =>
{
await HandleMsgAsync(webSocket,msgQueueUtil, authorizedHelper);
});
//Func<string, Task> SendAsync = async (text) =>
//{
// await WebSocketServer.SendAsync(webSocket, text);
//};
var buffer = new byte[1024];
var receivedMessage = new StringBuilder(); // 用于拼接长消息
while (webSocket.State == WebSocketState.Open)
while ( webSocket.State == WebSocketState.Open)
{
WebSocketReceiveResult result;
try
{
WebSocketReceiveResult result;
var buffer = new byte[1024];
do
{
result = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
if (result.MessageType == WebSocketMessageType.Close)
{
await webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", CancellationToken.None);
if (IsCheckToken)
{
AuthorizedClients.TryRemove(authorizedHelper.AddresPort, out var _);
}
}
// 将接收到的部分消息解码并拼接
var partialMessage = Encoding.UTF8.GetString(buffer, 0, result.Count);
receivedMessage.Append(partialMessage);
} while (!result.EndOfMessage); // 循环直到接收到完整的消息
// 完整消息已经接收到，准备处理
var message = receivedMessage.ToString();
if (result.MessageType == WebSocketMessageType.Close)
{
//SendAsync = null;
await webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", CancellationToken.None);
if (IsCheckToken)
{
AuthorizedClients.TryRemove(authorizedHelper.AddresPort, out var _);
}
}
else
{
if (IsCheckToken)
{
var authorizedResult = await authorizedHelper.HandleAuthorized(message); // 尝试检测授权
if (!authorizedResult) // 授权失败
{
await webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", CancellationToken.None);
if (IsCheckToken)
{
AuthorizedClients.TryRemove(authorizedHelper.AddresPort, out var _);
}
continue;
}
}
// 消息处理
_ = MsgHandleHelper.HandleMsgAsync(async (text) =>
{
await WebSocketServer.SendAsync(webSocket, text);
}, message); // 处理消息
}
// 清空 StringBuilder 为下一条消息做准备
receivedMessage.Clear();
var message = receivedMessage.ToString(); // 获取消息文本
receivedMessage.Clear(); // 清空 StringBuilder 为下一条消息做准备
msgQueueUtil.WriteMsg(message); // 处理消息
}
catch (Exception ex)
{
@@ -249,17 +228,45 @@ namespace Serein.Library.Network.WebSocketCommunication
}
}
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="webSocket"></param>
/// <param name="message"></param>
/// <returns></returns>
public static async Task SendAsync(WebSocket webSocket, string message)
public async Task HandleMsgAsync(WebSocket webSocket,
MsgQueueUtil msgQueueUtil,
WebSocketAuthorizedHelper authorizedHelper)
{
var buffer = Encoding.UTF8.GetBytes(message);
await webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
while (true)
{
var message = await msgQueueUtil.WaitMsgAsync(); // 有消息时通知
//if (!msgQueueUtil.TryGetMsg(out var message)) // 获取消息
//{
// return;
//}
if (IsCheckToken)
{
var authorizedResult = await authorizedHelper.HandleAuthorized(message); // 尝试检测授权
if (!authorizedResult) // 授权失败
{
await webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", CancellationToken.None);
if (IsCheckToken)
{
AuthorizedClients.TryRemove(authorizedHelper.AddresPort, out var _);
}
return;
}
}
// 消息处理
MsgHandleHelper.HandleMsg(async (text) =>
{
await SocketExtension.SendAsync(webSocket, text); // 回复客户端，处理方法中入参如果需要发送消息委托，则将该回调方法作为委托参数传入
}, message); // 处理消息
}
}
}
}