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

修改了无法添加基础节点的bug，增加WebSocket JSON ID字段，远程环境交互使用消息ID作为响应key。

Browse Source
This commit is contained in:
fengjiayi
2024-10-22 00:13:13 +08:00
parent 838158f446
commit 0a7e24d318
48 changed files with 1209 additions and 500 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
Library/Network/WebSocket/Attribute.cs
View File

@@ -27,6 +27,7 @@ namespace Serein.Library.Network.WebSocketCommunication
{
public string ThemeKey;
public string DataKey;
public string MsgIdKey;
}
@@ -71,13 +72,19 @@ namespace Serein.Library.Network.WebSocketCommunication
}
/// <summary>
/// 使用消息DataKey整体数据
/// 使用 DataKey 整体数据
/// </summary>
[AttributeUsage(AttributeTargets.Parameter)]
public sealed class UseMsgDataAttribute : Attribute
public sealed class UseDataAttribute : Attribute
{
}
/// <summary>
/// 使用 MsgIdKey 整体数据
/// </summary>
[AttributeUsage(AttributeTargets.Parameter)]
public sealed class UseMsgIdAttribute : Attribute
{
}
internal class SocketHandleModule
{

52
Library/Network/WebSocket/Handle/Attribute.cs
View File

@@ -2,6 +2,7 @@
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace Serein.Library.Network.WebSocketCommunication.Handle
@@ -12,4 +13,55 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
public sealed class NeedfulAttribute : Attribute
{
}
/// <summary>
/// 消息ID生成器
/// </summary>
public class MsgIdHelper
{
private static readonly long _epoch = new DateTime(2023, 1, 1).Ticks; // 自定义起始时间
private static long _lastTimestamp = -1L; // 上一次生成 ID 的时间戳
private static long _sequence = 0L; // 序列号
/// <summary>
/// 获取新的ID
/// </summary>
public static long NewId => GenerateId();
/// <summary>
/// 生成消息ID
/// </summary>
/// <returns></returns>
public static long GenerateId()
{
long timestamp = DateTime.UtcNow.Ticks;
// 如果时间戳是一样的，递增序列号
if (timestamp == _lastTimestamp)
{
// 使用原子操作增加序列号
_sequence = Interlocked.Increment(ref _sequence);
if (_sequence > 999999) // 序列号最大值，6位
{
// 等待下一毫秒
while (timestamp <= _lastTimestamp)
{
timestamp = DateTime.UtcNow.Ticks;
}
}
}
else
{
_sequence = 0; // 重置序列号
}
_lastTimestamp = timestamp;
// 生成 ID：时间戳和序列号拼接
return (timestamp - _epoch) * 1000 + _sequence; // 返回 ID
}
}
}

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

@@ -41,7 +41,8 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
this.OnExceptionTracking = onExceptionTracking;
this.ArgNotNull = ArgNotNull;
this.useMsgData = parameterInfos.Select(p => p.GetCustomAttribute<UseMsgDataAttribute>() != null).ToArray();
this.useData = parameterInfos.Select(p => p.GetCustomAttribute<UseDataAttribute>() != null).ToArray();
this.useMsgId = parameterInfos.Select(p => p.GetCustomAttribute<UseMsgIdAttribute>() != null).ToArray();
#if NET5_0_OR_GREATER
this.IsCheckArgNotNull = parameterInfos.Select(p => p.GetCustomAttribute<NotNullAttribute>() != null).ToArray();
#endif
@@ -101,17 +102,24 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
/// </summary>
private readonly Type[] ParameterType;
/// <summary>
/// 是否使用整体data参数
/// 是否使Data整体内容作为入参参数
/// </summary>
private readonly bool[] useMsgData;
private readonly bool[] useData;
/// <summary>
/// 是否使用消息ID作为入参参数
/// </summary>
private readonly bool[] useMsgId;
/// <summary>
/// 是否检查变量为空
/// </summary>
private readonly bool[] IsCheckArgNotNull;
//private object ConvertArg(Type type, string argName )
//{
//}
public async void Handle(Func<object, Task> SendAsync, JObject jsonObject)
public async void Handle(Func<object, Task> SendAsync,string msgId, JObject jsonObject)
{
object[] args = new object[ParameterType.Length];
bool isCanInvoke = true;; // 表示是否可以调用方法
@@ -119,10 +127,17 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
{
var type = ParameterType[i];
var argName = ParameterName[i];
if (useMsgData[i])
#region DATA JSON数据
if (useData[i])
{
args[i] = jsonObject.ToObject(type);
}
#endregion
else if (useMsgId[i])
{
args[i] = msgId;
}
#region 值类型参数
else if (type.IsValueType)
{
var jsonValue = jsonObject.GetValue(argName);
@@ -143,7 +158,9 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
break;
}
}
}
}
#endregion
#region 引用类型参数
else if (type.IsClass)
{
var jsonValue = jsonObject.GetValue(argName);
@@ -164,7 +181,9 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
break;
}
}
}
}
#endregion
#region 传递消息委托
else if (type.IsGenericType) // 传递SendAsync委托
{
if (type.IsAssignableFrom(typeof(Func<object, Task>)))
@@ -198,7 +217,8 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
await SendAsync.Invoke(jsonText);
});
}
}
}
#endregion
}
if (!isCanInvoke)
@@ -212,8 +232,6 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
if (EmitMethodType == EmitHelper.EmitMethodType.HasResultTask && EmitDelegate is Func<object, object[], Task<object>> hasResultTask)
{
result = await hasResultTask(Instance, args);
//Console.WriteLine(result);
// why not data?
}
else if (EmitMethodType == EmitHelper.EmitMethodType.Task && EmitDelegate is Func<object, object[], Task> task)
{
@@ -233,34 +251,30 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
{
result = null;
await Console.Out.WriteLineAsync(ex.Message);
this.OnExceptionTracking.Invoke(ex, (async data =>
this.OnExceptionTracking.Invoke(ex, (async exData =>
{
var jsonText = JsonConvert.SerializeObject(data);
await SendAsync.Invoke(jsonText);
await SendAsync.Invoke(exData);
}));
}
//sw.Stop();
//Console.WriteLine($"Emit Invoke：{sw.ElapsedTicks * 1000000F / Stopwatch.Frequency:n3}μs");
if(result is null)
if (Module.IsReturnValue)
{
return;
}
else
{
if (Module.IsReturnValue)
if (result is null)
{
_ = SendAsync.Invoke(result);
result = "null";
}
_ = SendAsync.Invoke(result);
}
//if( && result != null && result.GetType().IsClass)
//{
// //var reusltJsonText = JsonConvert.SerializeObject(result);
// //_ = SendAsync.Invoke($"{reusltJsonText}");
//}
}

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

@@ -19,21 +19,27 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
/// <summary>
/// Json消息处理模块
/// </summary>
public WebSocketHandleModule(string ThemeJsonKey, string DataJsonKey)
public WebSocketHandleModule(string themeJsonKey, string dataJsonKey, string msgIdJsonKey)
{
this.ThemeJsonKey = ThemeJsonKey;
this.DataJsonKey = DataJsonKey;
this.ThemeJsonKey = themeJsonKey;
this.DataJsonKey = dataJsonKey;
this.MsgIdJsonKey = msgIdJsonKey;
}
/// <summary>
/// 指示处理模块该使用json中的哪个key作为业务区别字段
/// 指示处理模块该使用 Json 中的哪个 Key 作为业务区别字段
/// </summary>
public string ThemeJsonKey { get; }
/// <summary>
/// 指示处理模块该使用json中的哪个key作为业务数据字段
/// 指示处理模块该使用 Json 中的哪个 Key 作为业务数据字段
/// </summary>
public string DataJsonKey { get; }
/// <summary>
/// 指示处理模块该使用 Json 中的哪个 Key 作为业务消息ID字段
/// </summary>
public string MsgIdJsonKey { get; }
/// <summary>
/// 存储处理数据的配置
@@ -88,34 +94,28 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
/// <summary>
/// 处理JSON数据
/// </summary>
/// <param name="tSendAsync"></param>
/// <param name="sendAsync"></param>
/// <param name="jsonObject"></param>
public void HandleSocketMsg(Func<string, Task> tSendAsync, JObject jsonObject)
public void HandleSocketMsg(Func<string, Task> sendAsync, JObject jsonObject)
{
// 获取到消息
string themeKeyName = jsonObject.GetValue(ThemeJsonKey)?.ToString();
if (!MyHandleConfigs.TryGetValue(themeKeyName, out var handldConfig))
string theme = jsonObject.GetValue(ThemeJsonKey)?.ToString();
if (!MyHandleConfigs.TryGetValue(theme, out var handldConfig))
{
// 没有主题
return;
}
string msgId = jsonObject.GetValue(MsgIdJsonKey)?.ToString();
Func<object, Task> SendAsync = async (data) =>
{
var sendMsg = new
{
theme = themeKeyName,
token = "",
data = data,
};
var msg = JsonConvert.SerializeObject(sendMsg);
await tSendAsync(msg);
};
try
{
JObject dataObj = jsonObject.GetValue(DataJsonKey).ToObject<JObject>();
handldConfig.Handle(SendAsync, dataObj);
JObject dataObj = jsonObject.GetValue(DataJsonKey)?.ToObject<JObject>();
handldConfig.Handle(async (data) =>
{
await this.SendAsync(sendAsync, msgId, theme, data);
}, msgId, dataObj);
}
catch (Exception ex)
@@ -123,12 +123,64 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
Console.WriteLine($"error in ws : {ex.Message}{Environment.NewLine}json value:{jsonObject}");
return;
}
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="sendAsync"></param>
/// <param name="msgId"></param>
/// <param name="theme"></param>
/// <param name="data"></param>
/// <returns></returns>
public async Task SendAsync(Func<string, Task> sendAsync,string msgId, string theme, object data)
{
JObject jsonData;
if (data is null)
{
jsonData = new JObject()
{
[MsgIdJsonKey] = msgId,
[ThemeJsonKey] = theme,
};
}
else
{
JToken dataToken;
if ((data is System.Collections.IEnumerable || data is Array))
{
dataToken = JArray.FromObject(data);
}
else
{
dataToken = JObject.FromObject(data);
}
jsonData = new JObject()
{
[MsgIdJsonKey] = msgId,
[ThemeJsonKey] = theme,
[DataJsonKey] = dataToken
};
}
var msg = jsonData.ToString();
//Console.WriteLine(msg);
//Console.WriteLine();
await sendAsync.Invoke(msg);
}
}
}

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

@@ -41,13 +41,14 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
/// </summary>
/// <param name="themeKeyName"></param>
/// <param name="dataKeyName"></param>
/// <param name="msgIdKeyName"></param>
/// <returns></returns>
private WebSocketHandleModule AddMyHandleModule(string themeKeyName, string dataKeyName)
private WebSocketHandleModule AddMyHandleModule(string themeKeyName, string dataKeyName, string msgIdKeyName)
{
var key = (themeKeyName, dataKeyName);
if (!MyHandleModuleDict.TryGetValue(key, out var myHandleModule))
{
myHandleModule = new WebSocketHandleModule(themeKeyName, dataKeyName);
myHandleModule = new WebSocketHandleModule(themeKeyName, dataKeyName, msgIdKeyName);
MyHandleModuleDict[key] = myHandleModule;
}
return myHandleModule;
@@ -93,8 +94,9 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
var themeKey = moduleAttribute.ThemeKey;
var dataKey = moduleAttribute.DataKey;
var msgIdKey = moduleAttribute.MsgIdKey;
var handlemodule = AddMyHandleModule(themeKey, dataKey);
var handleModule = AddMyHandleModule(themeKey, dataKey, msgIdKey);
var methods = type.GetMethods(BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic)
.Select(method =>
{
@@ -135,7 +137,7 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
try
{
var jsonMsgHandleConfig = new JsonMsgHandleConfig(module, socketControlBase, method, onExceptionTracking, argNotNull);
var result = handlemodule.AddHandleConfigs(module,jsonMsgHandleConfig);
var result = handleModule.AddHandleConfigs(module,jsonMsgHandleConfig);
if (!result)
{
throw new Exception("添加失败，已经添加过相同的配置");
@@ -158,6 +160,7 @@ namespace Serein.Library.Network.WebSocketCommunication.Handle
/// <returns></returns>
public async Task HandleMsgAsync(Func<string, Task> SendAsync, string message)
{
//Console.WriteLine(message);
JObject json = JObject.Parse(message);
await Task.Run(() =>
{

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

@@ -96,7 +96,7 @@ namespace Serein.Library.Network.WebSocketCommunication
{
var completeMessage = receivedMessage.ToString();
_ = MsgHandleHelper.HandleMsgAsync(SendAsync, completeMessage); // 处理消息
Debug.WriteLine($"Received: {completeMessage}");
//Debug.WriteLine($"Received: {completeMessage}");
}
// 清空 StringBuilder 为下一条消息做准备

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

@@ -180,10 +180,10 @@ namespace Serein.Library.Network.WebSocketCommunication
return;
}
Func<string, Task> SendAsync = async (text) =>
{
await WebSocketServer.SendAsync(webSocket, text);
};
//Func<string, Task> SendAsync = async (text) =>
//{
// await WebSocketServer.SendAsync(webSocket, text);
//};
var buffer = new byte[1024];
var receivedMessage = new StringBuilder(); // 用于拼接长消息
@@ -209,7 +209,7 @@ namespace Serein.Library.Network.WebSocketCommunication
if (result.MessageType == WebSocketMessageType.Close)
{
SendAsync = null;
//SendAsync = null;
await webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", CancellationToken.None);
if (IsCheckToken)
{
@@ -233,7 +233,10 @@ namespace Serein.Library.Network.WebSocketCommunication
}
// 消息处理
_ = MsgHandleHelper.HandleMsgAsync(SendAsync, message); // 处理消息
_ = MsgHandleHelper.HandleMsgAsync(async (text) =>
{
await WebSocketServer.SendAsync(webSocket, text);
}, message); // 处理消息
}
// 清空 StringBuilder 为下一条消息做准备