Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
116cbcdc7f | ||
|
|
4e1b7d4845 | ||
|
|
71acb35df9 | ||
|
|
67cc7346dc | ||
|
|
8627ae61fa | ||
|
|
02523028db | ||
|
|
7d39ab5e0b |
@@ -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`
|
||||||
|
|||||||
@@ -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
@@ -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>
|
||||||
|
|||||||
@@ -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
@@ -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();
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
@@ -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();
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user