1. RpcServer can return not found , 2. Event server process all parllels

This commit is contained in:
2024-08-05 19:55:14 +08:00
parent a4bae0226e
commit 141846d79d
2 changed files with 63 additions and 55 deletions
+12 -10
View File
@@ -1,4 +1,5 @@
using System; using System;
using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Text.Json; using System.Text.Json;
using System.Text.RegularExpressions; using System.Text.RegularExpressions;
@@ -12,14 +13,13 @@ namespace SimApi;
public partial class Synapse public partial class Synapse
{ {
private string EventServerTopicPrefix => $"{Options.SysName}/event/"; private string EventServerTopicPrefix => $"{Options.SysName}/event/";
private void RunEventServer() private void RunEventServer()
{ {
Client!.ApplicationMessageReceivedAsync += e => Client!.ApplicationMessageReceivedAsync += async e =>
{ {
if (!e.ApplicationMessage.Topic.StartsWith(EventServerTopicPrefix)) return Task.CompletedTask; if (!e.ApplicationMessage.Topic.StartsWith(EventServerTopicPrefix)) return;
var reqBody = e.ApplicationMessage.ConvertPayloadToString(); var reqBody = e.ApplicationMessage.ConvertPayloadToString();
var eventName = e.ApplicationMessage.Topic.Replace(EventServerTopicPrefix, string.Empty); var eventName = e.ApplicationMessage.Topic.Replace(EventServerTopicPrefix, string.Empty);
logger.LogDebug("Synapse Event Receive: {EventName}\n{Body}", eventName, reqBody); logger.LogDebug("Synapse Event Receive: {EventName}\n{Body}", eventName, reqBody);
@@ -27,7 +27,7 @@ public partial class Synapse
.Where(x => Regex.IsMatch(eventName, .Where(x => Regex.IsMatch(eventName,
"^" + Regex.Escape(x.Key!).Replace("\\+", "[^/]+").Replace("\\#", ".*") + "$")) "^" + Regex.Escape(x.Key!).Replace("\\+", "[^/]+").Replace("\\#", ".*") + "$"))
.ToArray(); .ToArray();
foreach (var method in methods) var tasks = methods.Select(method => Task.Run(() =>
{ {
var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(method.Class!); var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(method.Class!);
var mt = callClass.GetType().GetMethod(method.Method!); var mt = callClass.GetType().GetMethod(method.Method!);
@@ -37,20 +37,21 @@ public partial class Synapse
{ {
var pt = mt.GetParameters()[1].ParameterType; var pt = mt.GetParameters()[1].ParameterType;
mt.Invoke(callClass, pt == typeof(string) mt.Invoke(callClass, pt == typeof(string)
? [eventName, reqBody] ? new object?[] { eventName, reqBody }
: [eventName, JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption)]); : new object?[] { eventName, JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption) });
} }
else else
{ {
mt.Invoke(callClass, [eventName]); mt.Invoke(callClass, new object[] { eventName });
} }
} }
catch (Exception ex) catch (Exception ex)
{ {
logger.LogError("Synapse Event Processor Error: {Err}\n{Stack}", ex.Message,ex.StackTrace); logger.LogError("Synapse Event Processor Error: {Err}\n{Stack}", ex.Message, ex.StackTrace);
} }
} }))
return Task.CompletedTask; .ToList();
await Task.WhenAll(tasks);
}; };
SubEventServerTopic(); SubEventServerTopic();
} }
@@ -64,6 +65,7 @@ public partial class Synapse
{ {
topic = "$queue/" + topic; topic = "$queue/" + topic;
} }
var evSubOpts = MqttFactory.CreateSubscribeOptionsBuilder() var evSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter(o => o.WithTopic(topic)).Build(); .WithTopicFilter(o => o.WithTopic(topic)).Build();
Client!.SubscribeAsync(evSubOpts).Wait(); Client!.SubscribeAsync(evSubOpts).Wait();
+7 -1
View File
@@ -32,7 +32,12 @@ public partial class Synapse
reqBody); reqBody);
SimApiBaseResponse res; SimApiBaseResponse res;
var method = RpcRegistry.FirstOrDefault(x => x.Key == action); var method = RpcRegistry.FirstOrDefault(x => x.Key == action);
if (method == null) return Task.CompletedTask; if (method == null)
{
res = new SimApiBaseResponse(404, "method not found");
}
else
{
var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(method.Class!); var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(method.Class!);
var mt = callClass.GetType().GetMethod(method.Method!); var mt = callClass.GetType().GetMethod(method.Method!);
try try
@@ -74,6 +79,7 @@ public partial class Synapse
logger.LogDebug("Synapse RPC调用失败: {Err}", ex.Message); logger.LogDebug("Synapse RPC调用失败: {Err}", ex.Message);
res = new SimApiBaseResponse(500, 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]}/{e.ApplicationMessage.ContentType}"; var reply = $"{Options.SysName}/{appInfo[0]}/rpc/client/{appInfo[1]}/{e.ApplicationMessage.ContentType}";