Compare commits

...
7 Commits
Author SHA1 Message Date
xrain 116cbcdc7f update to 9 2025-10-14 02:55:29 +08:00
xrain 4e1b7d4845 update to 9 2025-10-14 02:20:37 +08:00
xrain 71acb35df9 cancle json refrence 2025-08-13 17:28:24 +08:00
xrain 67cc7346dc use concurrentDic 2025-05-17 21:17:43 +08:00
xrain 8627ae61fa use lock 2025-05-17 21:11:27 +08:00
xrain 02523028db add task.run 2025-05-17 20:40:48 +08:00
xrain 7d39ab5e0b add task.run 2025-05-17 20:38:39 +08:00
7 changed files with 117 additions and 112 deletions
+1 -1
View File
@@ -14,7 +14,7 @@ jobs:
- name: Setup .NET Core - name: Setup .NET Core
uses: actions/setup-dotnet@v1 uses: actions/setup-dotnet@v1
with: with:
dotnet-version: "8.0.200" dotnet-version: "9.0.305"
- name: Publish - name: Publish
run: | run: |
version=`git describe --tags` version=`git describe --tags`
+2 -2
View File
@@ -24,10 +24,10 @@ public static class SimApiUtil
/// </summary> /// </summary>
public static JsonSerializerOptions JsonOption => new() public static JsonSerializerOptions JsonOption => new()
{ {
ReferenceHandler = ReferenceHandler.Preserve, // ReferenceHandler = ReferenceHandler.Preserve,
Encoder = JavaScriptEncoder.Create(UnicodeRanges.All), Encoder = JavaScriptEncoder.Create(UnicodeRanges.All),
PropertyNamingPolicy = JsonNamingPolicy.CamelCase, PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull // DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull
}; };
/// <summary> /// <summary>
+1 -1
View File
@@ -14,8 +14,8 @@
<SynchReleaseVersion>false</SynchReleaseVersion> <SynchReleaseVersion>false</SynchReleaseVersion>
<GeneratePackageOnBuild>true</GeneratePackageOnBuild> <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
<PackageVersion>5.0.2</PackageVersion> <PackageVersion>5.0.2</PackageVersion>
<TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable> <Nullable>enable</Nullable>
<TargetFrameworks>net8.0;net9.0</TargetFrameworks>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
-2
View File
@@ -3,7 +3,6 @@ using System.Diagnostics;
using System.Linq; using System.Linq;
using System.Reflection; using System.Reflection;
using System.Text.Json; using System.Text.Json;
using System.Text.Json.Serialization;
using Hangfire; using Hangfire;
using Hangfire.Console; using Hangfire.Console;
using Hangfire.Redis.StackExchange; using Hangfire.Redis.StackExchange;
@@ -251,7 +250,6 @@ public static class SimApiExtensions
.AddJsonOptions(opt => .AddJsonOptions(opt =>
{ {
opt.JsonSerializerOptions.PropertyNamingPolicy = JsonNamingPolicy.CamelCase; opt.JsonSerializerOptions.PropertyNamingPolicy = JsonNamingPolicy.CamelCase;
opt.JsonSerializerOptions.ReferenceHandler = ReferenceHandler.Preserve;
}); });
} }
+31 -28
View File
@@ -19,39 +19,42 @@ public partial class Synapse
{ {
Client!.ApplicationMessageReceivedAsync += async e => Client!.ApplicationMessageReceivedAsync += async e =>
{ {
if (!e.ApplicationMessage.Topic.StartsWith(EventServerTopicPrefix)) return; await Task.Run(async () =>
var reqBody = e.ApplicationMessage.ConvertPayloadToString(); {
var eventName = e.ApplicationMessage.Topic.Replace(EventServerTopicPrefix, string.Empty); if (!e.ApplicationMessage.Topic.StartsWith(EventServerTopicPrefix)) return;
logger.LogDebug("Synapse Event Receive: {EventName}\n{Body}", eventName, reqBody); var reqBody = e.ApplicationMessage.ConvertPayloadToString();
var methods = EventRegistry var eventName = e.ApplicationMessage.Topic.Replace(EventServerTopicPrefix, string.Empty);
.Where(x => Regex.IsMatch(eventName, logger.LogDebug("Synapse Event Receive: {EventName}\n{Body}", eventName, reqBody);
"^" + Regex.Escape(x.Key!).Replace("\\+", "[^/]+").Replace("\\#", ".*") + "$")) var methods = EventRegistry
.ToArray(); .Where(x => Regex.IsMatch(eventName,
var tasks = methods.Select(method => Task.Run(() => "^" + Regex.Escape(x.Key!).Replace("\\+", "[^/]+").Replace("\\#", ".*") + "$"))
{ .ToArray();
var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(method.Class!); var tasks = methods.Select(method => Task.Run(() =>
var mt = callClass.GetType().GetMethod(method.Method!);
try
{ {
if (mt!.GetParameters().Length == 2) var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(method.Class!);
var mt = callClass.GetType().GetMethod(method.Method!);
try
{ {
var pt = mt.GetParameters()[1].ParameterType; if (mt!.GetParameters().Length == 2)
mt.Invoke(callClass, pt == typeof(string) {
? new object?[] { eventName, reqBody } var pt = mt.GetParameters()[1].ParameterType;
: new object?[] { eventName, JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption) }); mt.Invoke(callClass, pt == typeof(string)
? [eventName, reqBody]
: [eventName, JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption)]);
}
else
{
mt.Invoke(callClass, new object[] { eventName });
}
} }
else catch (Exception ex)
{ {
mt.Invoke(callClass, new object[] { eventName }); logger.LogError("Synapse Event Processor Error: {Err}\n{Stack}", ex.Message, ex.StackTrace);
} }
} }))
catch (Exception ex) .ToList();
{ await Task.WhenAll(tasks);
logger.LogError("Synapse Event Processor Error: {Err}\n{Stack}", ex.Message, ex.StackTrace); });
}
}))
.ToList();
await Task.WhenAll(tasks);
}; };
SubEventServerTopic(); SubEventServerTopic();
} }
+5 -5
View File
@@ -1,20 +1,19 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq;
using System.Text.Json; using System.Text.Json;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using MQTTnet; using MQTTnet;
using MQTTnet.Packets;
using SimApi.Communications; using SimApi.Communications;
using SimApi.Helpers; using SimApi.Helpers;
using System.Collections.Concurrent;
namespace SimApi; namespace SimApi;
public partial class Synapse public partial class Synapse
{ {
private Dictionary<string, TaskCompletionSource<string>> ResponseCompletionSources { get; } = new(); private ConcurrentDictionary<string, TaskCompletionSource<string>> ResponseCompletionSources { get; } = new();
private string EventClientTopicPrefix => $"{Options.SysName}/{Options.AppName}/rpc/client/{Options.AppId}/"; private string EventClientTopicPrefix => $"{Options.SysName}/{Options.AppName}/rpc/client/{Options.AppId}/";
@@ -27,7 +26,7 @@ public partial class Synapse
var messageId = e.ApplicationMessage.Topic.Replace(EventClientTopicPrefix, string.Empty); var messageId = e.ApplicationMessage.Topic.Replace(EventClientTopicPrefix, string.Empty);
if (!ResponseCompletionSources.TryGetValue(messageId, out var tcs)) return Task.CompletedTask; if (!ResponseCompletionSources.TryGetValue(messageId, out var tcs)) return Task.CompletedTask;
tcs.SetResult(reqBody); tcs.SetResult(reqBody);
ResponseCompletionSources.Remove(messageId); ResponseCompletionSources.Remove(messageId, out _);
return Task.CompletedTask; return Task.CompletedTask;
}; };
SubRpcClientTopic(); SubRpcClientTopic();
@@ -43,6 +42,7 @@ public partial class Synapse
private string? FireRpc(string app, string action, object? param, Dictionary<string, string>? headers = null, private string? FireRpc(string app, string action, object? param, Dictionary<string, string>? headers = null,
int? timeout = null) int? timeout = null)
{ {
// 移除锁语句
string paramJson; string paramJson;
if (param is string strParam) if (param is string strParam)
{ {
@@ -56,7 +56,7 @@ public partial class Synapse
var topic = $"{Options.SysName}/{app}/rpc/server/{action}"; var topic = $"{Options.SysName}/{app}/rpc/server/{action}";
var messageId = Guid.NewGuid().ToString(); var messageId = Guid.NewGuid().ToString();
var tcs = new TaskCompletionSource<string>(); var tcs = new TaskCompletionSource<string>();
ResponseCompletionSources.Add(messageId, tcs); ResponseCompletionSources.TryAdd(messageId, tcs);
var messageBuilder = new MqttApplicationMessageBuilder() var messageBuilder = new MqttApplicationMessageBuilder()
.WithTopic(topic) .WithTopic(topic)
.WithPayload(paramJson) .WithPayload(paramJson)
+77 -73
View File
@@ -20,91 +20,95 @@ public partial class Synapse
private void RunRpcServer() private void RunRpcServer()
{ {
Client!.ApplicationMessageReceivedAsync += e => Client!.ApplicationMessageReceivedAsync += async e =>
{ {
if (!e.ApplicationMessage.Topic.StartsWith(RpcServerTopicPrefix)) return Task.CompletedTask; await Task.Run(() =>
var reqBody = e.ApplicationMessage.ConvertPayloadToString();
var action = e.ApplicationMessage.Topic.Replace(RpcServerTopicPrefix, string.Empty);
var appInfo = e.ApplicationMessage.ResponseTopic.Split(",");
logger.LogDebug(
"Synapse RPC Server Receive: ({BasicPropertiesMessageId}) {BasicPropertiesReplyTo} -> {BasicPropertiesType}@{OptionsAppName}\n{S}",
appInfo[2], appInfo[0], action, Options.AppName, reqBody);
SimApiBaseResponse res;
var method = RpcRegistry.FirstOrDefault(x => x.Key == action);
if (method == null)
{ {
res = new SimApiBaseResponse(404, "method not found"); if (!e.ApplicationMessage.Topic.StartsWith(RpcServerTopicPrefix)) return;
} var reqBody = e.ApplicationMessage.ConvertPayloadToString();
else var action = e.ApplicationMessage.Topic.Replace(RpcServerTopicPrefix, string.Empty);
{ var appInfo = e.ApplicationMessage.ResponseTopic.Split(",");
var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(method.Class!); logger.LogDebug(
var mt = callClass.GetType().GetMethod(method.Method!); "Synapse RPC Server Receive: ({BasicPropertiesMessageId}) {BasicPropertiesReplyTo} -> {BasicPropertiesType}@{OptionsAppName}\n{S}",
try appInfo[2], appInfo[0], action, Options.AppName, reqBody);
SimApiBaseResponse res;
var method = RpcRegistry.FirstOrDefault(x => x.Key == action);
if (method == null)
{ {
var methodParams = mt!.GetParameters(); res = new SimApiBaseResponse(404, "method not found");
object? ret;
switch (methodParams.Length)
{
case 1:
var pt = mt.GetParameters()[0].ParameterType;
var param = pt == typeof(string)
? [reqBody]
: new[] { JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption) };
ret = mt.Invoke(callClass, param);
break;
case 2:
var headerData =
e.ApplicationMessage.UserProperties.ToDictionary(x => x.Name, x => x.Value);
var pt2 = mt.GetParameters()[0].ParameterType;
var param2 = pt2 == typeof(string)
? [reqBody]
: new[] { JsonSerializer.Deserialize(reqBody, pt2, SimApiUtil.JsonOption), headerData };
ret = mt.Invoke(callClass, param2);
break;
default:
ret = mt.Invoke(callClass, []);
break;
}
res = new SimApiBaseResponse<object?>
{
Data = ret
};
} }
catch (TargetInvocationException ex) else
{ {
if (ex.InnerException is SimApiException ie) var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(method.Class!);
var mt = callClass.GetType().GetMethod(method.Method!);
try
{ {
logger.LogDebug("Synapse RPC调用错误: {Err}", ie.Message); var methodParams = mt!.GetParameters();
res = new SimApiBaseResponse(ie.Code, ie.Message); object? ret;
switch (methodParams.Length)
{
case 1:
var pt = mt.GetParameters()[0].ParameterType;
var param = pt == typeof(string)
? [reqBody]
: new[] { JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption) };
ret = mt.Invoke(callClass, param);
break;
case 2:
var headerData =
e.ApplicationMessage.UserProperties.ToDictionary(x => x.Name, x => x.Value);
var pt2 = mt.GetParameters()[0].ParameterType;
var param2 = pt2 == typeof(string)
? [reqBody]
: new[]
{
JsonSerializer.Deserialize(reqBody, pt2, SimApiUtil.JsonOption), headerData
};
ret = mt.Invoke(callClass, param2);
break;
default:
ret = mt.Invoke(callClass, []);
break;
}
res = new SimApiBaseResponse<object?>
{
Data = ret
};
} }
else catch (TargetInvocationException ex)
{ {
logger.LogError("Synapse RPC 方法异常: {Err}\n{Stack}", ex.Message, ex.StackTrace); if (ex.InnerException is SimApiException ie)
{
logger.LogDebug("Synapse RPC调用错误: {Err}", ie.Message);
res = new SimApiBaseResponse(ie.Code, ie.Message);
}
else
{
logger.LogError("Synapse RPC 方法异常: {Err}\n{Stack}", ex.Message, ex.StackTrace);
res = new SimApiBaseResponse(500, ex.Message);
}
}
catch (Exception ex)
{
logger.LogDebug("Synapse RPC调用失败: {Err}", ex.Message);
res = new SimApiBaseResponse(500, ex.Message); res = new SimApiBaseResponse(500, ex.Message);
} }
} }
catch (Exception ex)
{
logger.LogDebug("Synapse RPC调用失败: {Err}", ex.Message);
res = new SimApiBaseResponse(500, ex.Message);
}
}
var returnJson = JsonSerializer.Serialize((object)res, SimApiUtil.JsonOption); var returnJson = JsonSerializer.Serialize((object)res, SimApiUtil.JsonOption);
var reply = $"{Options.SysName}/{appInfo[0]}/rpc/client/{appInfo[1]}/{appInfo[2]}"; var reply = $"{Options.SysName}/{appInfo[0]}/rpc/client/{appInfo[1]}/{appInfo[2]}";
var message = new MqttApplicationMessageBuilder() var message = new MqttApplicationMessageBuilder()
.WithTopic(reply) .WithTopic(reply)
.WithPayload(returnJson) .WithPayload(returnJson)
.WithRetainFlag(false) .WithRetainFlag(false)
.Build(); .Build();
if (!Client.IsConnected) return Task.CompletedTask; if (!Client.IsConnected) return;
Client.PublishAsync(message, CancellationToken.None).Wait(); Client.PublishAsync(message, CancellationToken.None).Wait();
logger.LogDebug( logger.LogDebug(
"Synapse Rpc Server Return: ({BasicPropertiesMessageId}) {BasicPropertiesType}@{OptionsAppName} -> {BasicPropertiesReplyTo}\n{ReturnJson}", "Synapse Rpc Server Return: ({BasicPropertiesMessageId}) {BasicPropertiesType}@{OptionsAppName} -> {BasicPropertiesReplyTo}\n{ReturnJson}",
appInfo[2], action, Options.AppName, appInfo[0], returnJson); appInfo[2], action, Options.AppName, appInfo[0], returnJson);
});
return Task.CompletedTask;
}; };
SubRpcServerTopic(); SubRpcServerTopic();
} }