zhusenlin/serein-flow
1
0
Fork 0
mirror of https://gitee.com/langsisi_admin/serein-flow synced 2026-03-02 15:50:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

优化了SereinWebSocket的API

Browse Source
This commit is contained in:
fengjiayi
2025-08-02 12:09:36 +08:00
parent 6fc57458a7
commit 93747ce7fd
10 changed files with 247 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
Library/Api/IJsonProvider.cs
View File

@@ -148,6 +148,14 @@ namespace Serein.Library.Api
/// <returns></returns>
IJsonToken Parse(string json);
/// <summary>
/// 尝试解析JSON文本为IJsonToken对象，如果成功则返回true，并通过out参数返回解析后的对象。
/// </summary>
/// <param name="json"></param>
/// <param name="jsonToken"></param>
/// <returns></returns>
bool TryParse(string json, out IJsonToken jsonToken);
/// <summary>
/// 创建对象
/// </summary>

13
Library/Utils/JsonHelper.cs
View File

@@ -58,6 +58,19 @@ namespace Serein.Library.Utils
return provider.Parse(json);
}
/// <summary>
/// 尝试解析Json文本为IJsonToken对象
/// </summary>
/// <param name="json"></param>
/// <param name="jsonToken"></param>
/// <returns></returns>
public static bool TryParse(string json, out IJsonToken jsonToken)
{
return provider.TryParse(json, out jsonToken);
}
/// <summary>
/// 将对象序列化为Json文本
/// </summary>

27
Serein.Extend.NewtonsoftJson/NewtonsoftJsonProvider.cs
View File

@@ -2,11 +2,12 @@
using Newtonsoft.Json.Linq;
using Newtonsoft.Json.Serialization;
using Serein.Library.Api;
using System.Diagnostics.CodeAnalysis;
namespace Serein.Extend.NewtonsoftJson
{
public enum JsonType
public enum ProviderType
{
Default = 0,
Web = 1,
@@ -30,11 +31,11 @@ namespace Serein.Extend.NewtonsoftJson
/// 基于Newtonsoft.Json的JSON门户实现
/// </summary>
/// <param name="jsonType"></param>
public NewtonsoftJsonProvider(JsonType jsonType)
public NewtonsoftJsonProvider(ProviderType jsonType)
{
settings = jsonType switch
{
JsonType.Web => new JsonSerializerSettings
ProviderType.Web => new JsonSerializerSettings
{
ContractResolver = new CamelCasePropertyNamesContractResolver(), // 控制首字母小写
NullValueHandling = NullValueHandling.Ignore // 可选：忽略 null
@@ -98,6 +99,26 @@ namespace Serein.Extend.NewtonsoftJson
return NewtonsoftJsonTokenFactory.Parse(json);
}
/// <summary>
/// 尝试解析JSON文本为IJsonToken对象，如果成功则返回true，并通过out参数返回解析后的对象。
/// </summary>
/// <param name="json"></param>
/// <param name="jsonToken"></param>
/// <returns></returns>
public bool TryParse(string json, [NotNullWhen(true)] out IJsonToken? jsonToken)
{
try
{
jsonToken = NewtonsoftJsonTokenFactory.Parse(json);
return true;
}
catch (Exception)
{
jsonToken = null!;
return false;
}
}
/// <summary>
/// 创建一个新的JSON数组对象。
/// </summary>

117
Serein.Proto.WebSocket/Handle/WebSocketHandleModule.cs
View File

@@ -91,80 +91,82 @@ namespace Serein.Proto.WebSocket.Handle
public async Task HandleAsync(WebSocketHandleContext context)
{
var jsonObject = context.MsgRequest; // 获取到消息
context.Model = new ModuleConfig(); // 设置当前模块配置
context.Model.IsResponseUseReturn = _moduleConfig.IsResponseUseReturn;
if (jsonObject is null)
{
context.TriggerExceptionTracking($"请求没有获取到消息");
return; // 没有获取到消息
}
if (!jsonObject.TryGetValue(_moduleConfig.MsgIdJsonKey, out var msgIdToken) || msgIdToken.IsNull)
{
context.TriggerExceptionTracking($"消息Id从JSON键[{_moduleConfig.MsgIdJsonKey}]提取失败");
return; // 没有获取到消息
}
if (msgIdToken.Type != IJsonToken.TokenType.Value)
{
context.TriggerExceptionTracking($"请求消息Id[{_moduleConfig.ThemeJsonKey}]需要值类型，当前类型为[{msgIdToken.Type}]");
return; // 没有获取到消息
}
var msgId = msgIdToken.ToString(); // 获取Id
context.Model.MsgId = msgId;
// 验证消息ID是否重复
if (!_myMsgIdHash.Add(msgId))
{
context.TriggerExceptionTracking($"消息Id[{msgId}]重复发送");
return; // 消息重复
}
if(!jsonObject.TryGetValue(_moduleConfig.ThemeJsonKey, out var themeToken) || themeToken.IsNull)
{
context.TriggerExceptionTracking($"请求没有获取到主题\"{_moduleConfig.ThemeJsonKey}\"");
context.TriggerExceptionTracking($"主题从JSON键[{_moduleConfig.ThemeJsonKey}]提取失败");
return; // 没有获取到消息
}
if(themeToken.Type != IJsonToken.TokenType.Value)
{
context.TriggerExceptionTracking($"请求主题需要值类型 \"{_moduleConfig.ThemeJsonKey}\"");
context.TriggerExceptionTracking($"请求主题[{_moduleConfig.ThemeJsonKey}]需要值类型，当前类型为[{themeToken.Type}]");
return; // 没有获取到消息
}
var theme = themeToken.ToString(); // 获取主题
context.Model.Theme = theme;
// 验证主题
if (!_methodInvokeConfigs.TryGetValue(theme, out var handldConfig))
{
context.TriggerExceptionTracking($"{_moduleConfig.ThemeJsonKey} 主题不存在");
context.TriggerExceptionTracking($"不存在这样的主题");
return;
}
if (!jsonObject.TryGetValue(_moduleConfig.MsgIdJsonKey, out var msgIdToken) || themeToken.IsNull)
{
context.TriggerExceptionTracking($"主题 {theme} 没有消息Id");
return; // 没有获取到消息
}
if (themeToken.Type != IJsonToken.TokenType.Value)
{
context.TriggerExceptionTracking($"请求消息Id需要值类型 \"{_moduleConfig.ThemeJsonKey}\"");
return; // 没有获取到消息
}
var msgId = msgIdToken.ToString(); // 获取主题
// 验证消息ID是否重复
if (!_myMsgIdHash.Add(msgId))
{
context.TriggerExceptionTracking($"主题 {theme} 消息Id {msgId} 消息重复");
return; // 消息重复
}
// 验证数据
if (!jsonObject.TryGetValue(_moduleConfig.DataJsonKey, out var dataToken))
{
context.TriggerExceptionTracking($"主题 {theme} 消息Id {msgId} 数据提取失败，当前指定键\"{_moduleConfig.DataJsonKey}\"");
context.TriggerExceptionTracking($"数据从JSON键[{_moduleConfig.DataJsonKey}]提取失败");
return; // 没有主题
}
if(dataToken.Type != IJsonToken.TokenType.Object)
{
context.TriggerExceptionTracking($"主题 {theme} 消息Id {msgId} 数据需要 JSON Object");
context.TriggerExceptionTracking($"数据需要 JSON Object，当前类型为[{dataToken.Type}]");
return;
}
context.MsgTheme = theme; // 添加主题
context.MsgId = msgId; // 添加 ID
context.MsgData = dataToken; // 添加消息
context.MsgRequest = jsonObject; // 添加原始消息
try
{
if (TryGetParameters(handldConfig, context, out var args))
{
var result = await HandleAsync(handldConfig, args);
var result = await InvokeAsync(handldConfig, args);
if (handldConfig.IsReturnValue)
{
await RepliedAsync(_moduleConfig, context, result);
await RepliedAsync(_moduleConfig, context, result);
}
}
else
{
context.TriggerExceptionTracking($"主题 {theme} 消息Id {msgId} 参数获取失败");
}
}
catch (Exception ex)
{
@@ -184,7 +186,7 @@ namespace Serein.Proto.WebSocket.Handle
/// <param name="config"></param>
/// <param name="args"></param>
/// <returns></returns>
public static async Task<object> HandleAsync(MethodInvokeConfiguration config, object?[] args)
public static async Task<object> InvokeAsync(MethodInvokeConfiguration config, object?[] args)
{
if (config.DelegateDetails is null)
{
@@ -206,8 +208,8 @@ namespace Serein.Proto.WebSocket.Handle
internal static bool TryGetParameters(MethodInvokeConfiguration config, WebSocketHandleContext context, out object?[] args)
{
args = new object[config.ParameterType.Length];
var theme = context.MsgTheme;
var msgId = context.MsgId;
var theme = context.Model.Theme;
var msgId = context.Model.MsgId;
List<string> exTips = [$"主题 {theme} 消息Id {msgId}"];
bool isCanInvoke = true; ; // 表示是否可以调用方法
for (int i = 0; i < config.ParameterType.Length; i++)
@@ -217,7 +219,7 @@ namespace Serein.Proto.WebSocket.Handle
#region 传递消息ID
if (config.UseMsgId[i])
{
args[i] = context.MsgId;
args[i] = msgId;
}
#endregion
#region DATA JSON数据
@@ -329,25 +331,40 @@ namespace Serein.Proto.WebSocket.Handle
WebSocketHandleContext context,
object data)
{
if (moduleConfig.IsResponseUseReturn)
if (context.OnExceptionTracking is null)
{
var responseContent = JsonHelper.Serialize(data);
await context.SendAsync(responseContent);
context.TriggerExceptionTracking(new NullReferenceException($"没有定义处理回复消息 OnExceptionTracking 回调函数"));
return;
}
else
// 返回结果
var responseData = context.OnReplyMakeData(context, data);
if (responseData is null)
{
IJsonToken jsonData;
jsonData = JsonHelper.Object(obj =>
{
obj[moduleConfig.MsgIdJsonKey] = context.MsgId;
obj[moduleConfig.ThemeJsonKey] = context.MsgTheme;
obj[moduleConfig.DataJsonKey] = data is null ? null : JsonHelper.FromObject(data);
});
var msg = jsonData.ToString();
await context.SendAsync(msg);
context.TriggerExceptionTracking(new ArgumentNullException($"处理回调函数 OnReplyMakeData 返回 null"));
return;
}
var responseContent = JsonHelper.Serialize(responseData);
await context.SendAsync(responseContent);
/* if (moduleConfig.IsResponseUseReturn)
{
var responseContent = JsonHelper.Serialize(data);
await context.SendAsync(responseContent);
}
else
{
IJsonToken jsonData;
jsonData = JsonHelper.Object(obj =>
{
obj[moduleConfig.MsgIdJsonKey] = context.MsgId;
obj[moduleConfig.ThemeJsonKey] = context.MsgTheme;
obj[moduleConfig.DataJsonKey] = data is null ? null : JsonHelper.FromObject(data);
});
var msg = jsonData.ToString();
await context.SendAsync(msg);
}*/
}
}

7
Serein.Proto.WebSocket/Handle/WebSocketModuleConfig.cs
View File

@@ -26,4 +26,11 @@
public bool IsResponseUseReturn { get; set; }
}
public class ModuleConfig()
{
public string MsgId { get; set; } = string.Empty;
public string Theme { get; set; } = string.Empty;
public bool IsResponseUseReturn { get; set; } = false;
}
}

18
Serein.Proto.WebSocket/ISereinWebSocketService.cs
View File

@@ -1,4 +1,5 @@
using System;
using Serein.Proto.WebSocket.Handle;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
@@ -34,9 +35,9 @@ namespace Serein.Proto.WebSocket
/// <summary>
/// 跟踪未处理的异常
/// </summary>
/// <param name="onExceptionTrackingAsync"></param>
/// <param name="onExceptionTracking"></param>
/// <returns></returns>
ISereinWebSocketService TrackUnhandledExceptions(Func<Exception, Func<object, Task>, Task> onExceptionTrackingAsync);
ISereinWebSocketService TrackUnhandledExceptions(Action<Exception> onExceptionTracking);
/// <summary>
/// 添加新的 WebSocket 连接进行处理消息
@@ -51,5 +52,16 @@ namespace Serein.Proto.WebSocket
/// <returns></returns>
Task PushDataAsync(object latestData);
/// <summary>
/// 设置回调函数，用于处理外部请求时的回复消息
/// </summary>
/// <param name="func"></param>
void OnReplyMakeData(Func<WebSocketHandleContext, object, object> func);
/// <summary>
/// 设置回调函数，回复外部请求时，记录消息内容
/// </summary>
/// <param name="onReply"></param>
void OnReply(Action<string> onReply);
}
}

34
Serein.Proto.WebSocket/SereinWebSocketHandleException.cs Normal file
View File

@@ -0,0 +1,34 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.Serialization;
using System.Text;
using System.Threading.Tasks;
namespace Serein.Proto.WebSocket
{
/// <summary>
/// WebSocket处理异常
/// </summary>
[Serializable]
internal sealed class SereinWebSocketHandleException : Exception
{
public SereinWebSocketHandleException()
{
}
public SereinWebSocketHandleException(string message)
: base(message)
{
}
public SereinWebSocketHandleException(string message, Exception innerException)
: base(message, innerException)
{
}
private SereinWebSocketHandleException(SerializationInfo info, StreamingContext context): base(info, context)
{
}
}
}

50
Serein.Proto.WebSocket/SereinWebSocketService.cs
View File

@@ -32,7 +32,11 @@ namespace Serein.Proto.WebSocket
/// 追踪未处理的异常
/// </summary>
private Func<Exception, Func<object, Task>, Task> _onExceptionTrackingAsync;
private Action<Exception>? _onExceptionTracking;
private Action<string>? _onReply;
private Func<WebSocketHandleContext, object, object> _onReplyMakeData;
/// <summary>
/// 维护所有 WebSocket 连接
/// </summary>
@@ -45,6 +49,9 @@ namespace Serein.Proto.WebSocket
public int ConcetionCount => _sockets.Count;
/// <summary>
/// SereinWebSocketService 构造函数，初始化 WebSocket 模块字典和连接列表
/// </summary>
public SereinWebSocketService()
{
_socketModules = new ConcurrentDictionary<(string, string), WebSocketHandleModule>();
@@ -53,9 +60,6 @@ namespace Serein.Proto.WebSocket
}
#region 添加处理模块
/// <summary>
@@ -247,14 +251,12 @@ namespace Serein.Proto.WebSocket
/// 跟踪未处理的异常
/// </summary>
/// <returns></returns>
public ISereinWebSocketService TrackUnhandledExceptions(Func<Exception, Func<object, Task>, Task> onExceptionTrackingAsync)
public ISereinWebSocketService TrackUnhandledExceptions(Action<Exception> onExceptionTracking)
{
_onExceptionTrackingAsync = onExceptionTrackingAsync;
_onExceptionTracking = onExceptionTracking;
return this;
}
/// <summary>
/// 传入新的 WebSocket 连接，开始进行处理
/// </summary>
@@ -280,6 +282,7 @@ namespace Serein.Proto.WebSocket
try
{
while (socket.State == WebSocketState.Open)
{
var result = await socket.ReceiveAsync(buffer: new ArraySegment<byte>(buffer), CancellationToken.None);
@@ -332,16 +335,23 @@ namespace Serein.Proto.WebSocket
while (webSocket.State == WebSocketState.Open)
{
var message = await tranTool.WaitMsgAsync(); // 有消息时通知
if (!JsonHelper.TryParse(message, out var jsonReques))
{
Console.WriteLine($"WebSocket 消息解析失败: {message}");
continue;
}
var context = new WebSocketHandleContext(sendasync);
context.MsgRequest = JsonHelper.Parse(message);
context.OnExceptionTrackingAsync = _onExceptionTrackingAsync;
context.MsgRequest = jsonReques;
context.OnExceptionTracking = _onExceptionTracking;
context.OnReplyMakeData = _onReplyMakeData;
context.OnReply = _onReply;
await HandleAsync(context); // 处理消息
}
}
/// <summary>
/// 异步消息
/// 异步处理消息
/// </summary>
/// <param name="context">此次请求的上下文</param>
/// <returns></returns>
@@ -395,6 +405,24 @@ namespace Serein.Proto.WebSocket
}
}
}
/// <summary>
/// 设置回调函数，用于处理外部请求时的回复消息
/// </summary>
/// <param name="func"></param>
public void OnReplyMakeData(Func<WebSocketHandleContext, object, object> func)
{
_onReplyMakeData = func;
}
/// <summary>
/// 设置回调函数，回复外部请求时，记录消息内容
/// </summary>
/// <param name="onReply"></param>
public void OnReply(Action<string> onReply)
{
_onReply = onReply;
}
}
}

6
Serein.Proto.WebSocket/TestClass.cs
View File

@@ -26,12 +26,8 @@ namespace Serein.Proto.WebSocket
SereinWebSocketService sereinWebSocketService = new SereinWebSocketService();
sereinWebSocketService.AddHandleModule<ClassA>();
sereinWebSocketService.AddHandleModule<ClassB>(() => new ClassB());
sereinWebSocketService.TrackUnhandledExceptions(OnExceptionTrackingAsync);
}
private static async Task OnExceptionTrackingAsync(Exception ex, Func<object, Task> SendAsync)
{
await SendAsync("");
}
}
}

67
Serein.Proto.WebSocket/WebSocketHandleContext.cs
View File

@@ -3,6 +3,7 @@ using Serein.Library.Utils;
using Serein.Proto.WebSocket.Handle;
using System;
using System.Threading.Tasks;
using static System.Runtime.InteropServices.JavaScript.JSType;
namespace Serein.Proto.WebSocket
{
@@ -27,8 +28,8 @@ namespace Serein.Proto.WebSocket
public void Dispose()
{
MsgRequest = null;
MsgTheme = string.Empty;
MsgId = string.Empty;
/*MsgTheme = string.Empty;
MsgId = string.Empty; */
MsgData = null;
MsgData = null;
}
@@ -46,22 +47,16 @@ namespace Serein.Proto.WebSocket
private bool _handle = false;
/// <summary>
/// WebSocket 模块配置
/// </summary>
public ModuleConfig Model { get; set; }
/// <summary>
/// 消息本体（IJsonToken）
/// </summary>
public IJsonToken? MsgRequest { get; set; }
/// <summary>
/// 此次消息请求的主题
/// </summary>
public string MsgTheme { get; set; } = string.Empty;
/// <summary>
/// 此次消息附带的ID
/// </summary>
public string MsgId { get; set; } = string.Empty;
/// <summary>
/// 此次消息的数据
/// </summary>
@@ -70,7 +65,23 @@ namespace Serein.Proto.WebSocket
/// <summary>
/// 异常外部感知使能
/// </summary>
public Func<Exception, Func<object, Task>, Task> OnExceptionTrackingAsync { get; set; }
public Action<Exception>? OnExceptionTracking { get; set; }
/// <summary>
/// 处理回复消息的函数
/// </summary>
public Func<WebSocketHandleContext, object, object> OnReplyMakeData { get; set; }
public Action<string>? OnReply { get; set; }
/// <summary>
/// 是否发生错误
/// </summary>
public bool IsError { get; set; }
/// <summary>
/// 错误消息
/// </summary>
public string ErrorMessage { get; set; }
/// <summary>
/// 发送消息
@@ -93,28 +104,28 @@ namespace Serein.Proto.WebSocket
/// </summary>
public void TriggerExceptionTracking(string exMessage)
{
var ex = new Exception(exMessage);
Func<object, Task> func = async (data) =>
{
var msg = JsonHelper.Serialize(data);
await _sendAsync.Invoke(msg);
};
OnExceptionTrackingAsync.Invoke(ex, func);
var ex = new SereinWebSocketHandleException(exMessage);
TriggerExceptionTracking(ex);
}
/// <summary>
/// 触发异常追踪
/// </summary>
public void TriggerExceptionTracking(Exception ex)
{
Func<object, Task> func = async (data) =>
IsError = true;
var msgId = Model.MsgId;
var theme = Model.Theme;
ErrorMessage = $"请求[{msgId}]主题[{theme}]异常 ：{ex.Message}";
OnExceptionTracking?.Invoke(ex);
var error = OnReplyMakeData?.Invoke(this, ex.Message); // 触发回复消息
_ = SendAsync(JsonHelper.Serialize(error)).ContinueWith((t) =>
{
var msg = JsonHelper.Serialize(data);
await _sendAsync.Invoke(msg);
};
OnExceptionTrackingAsync.Invoke(ex, func);
if (t.IsFaulted)
{
Console.WriteLine($"发送错误消息失败: {t.Exception?.Message}");
}
}); // 发送错误消息
}