add config store

This commit is contained in:
2024-07-26 08:11:21 +08:00
parent 82f7e0fc00
commit 929ddaced3
21 changed files with 302 additions and 153 deletions
+1 -1
View File
@@ -43,7 +43,7 @@ public class SimApiAuthAttribute : ActionFilterAttribute
public override void OnActionExecuting(ActionExecutingContext context) public override void OnActionExecuting(ActionExecutingContext context)
{ {
var loginInfo = (SimApiLoginItem)context.HttpContext.Items["LoginInfo"]; var loginInfo = (SimApiLoginItem)context.HttpContext.Items["LoginInfo"]!;
//检测是否登录 //检测是否登录
if (loginInfo == null) if (loginInfo == null)
{ {
+2 -7
View File
@@ -3,12 +3,7 @@ using System;
namespace SimApi.Attributes; namespace SimApi.Attributes;
[AttributeUsage(AttributeTargets.Method)] [AttributeUsage(AttributeTargets.Method)]
public class SynapseEventAttribute : Attribute public class SynapseEventAttribute(string? name = null) : Attribute
{ {
public string Name { get; } public string? Name { get; } = name;
public SynapseEventAttribute(string name)
{
Name = name;
}
} }
+1 -1
View File
@@ -5,7 +5,7 @@ namespace SimApi.Attributes;
[AttributeUsage(AttributeTargets.Method)] [AttributeUsage(AttributeTargets.Method)]
public class SynapseRpcAttribute : Attribute public class SynapseRpcAttribute : Attribute
{ {
public string Name { get; } public string? Name { get; }
public SynapseRpcAttribute() public SynapseRpcAttribute()
{ {
+2 -2
View File
@@ -15,7 +15,7 @@ public class SimApiIdOnlyRequest
/// </summary> /// </summary>
public class SimApiStringIdOnlyRequest public class SimApiStringIdOnlyRequest
{ {
[Required] public string Id { get; set; } [Required] public string? Id { get; set; }
} }
/// <summary> /// <summary>
@@ -24,7 +24,7 @@ public class SimApiStringIdOnlyRequest
/// <typeparam name="T"></typeparam> /// <typeparam name="T"></typeparam>
public class SimApiOneFieldRequest<T> public class SimApiOneFieldRequest<T>
{ {
[Required] public T Data { get; set; } [Required] public T? Data { get; set; }
} }
/// <summary> /// <summary>
+1 -1
View File
@@ -5,4 +5,4 @@ namespace SimApi.Communications;
/// <summary> /// <summary>
/// 登录信息中间件 /// 登录信息中间件
/// </summary> /// </summary>
public record SimApiLoginItem(string Id, string[] Type,Dictionary<string,string> Meta = null); public record SimApiLoginItem(string Id, string[] Type,Dictionary<string,string>? Meta = null);
+5 -5
View File
@@ -5,19 +5,19 @@ public class SimApiStorageOptions
/// <summary> /// <summary>
/// S3 服务器入口地址 /// S3 服务器入口地址
/// </summary> /// </summary>
public string Endpoint { get; set; } public string? Endpoint { get; set; }
/// <summary> /// <summary>
/// S3 服务器文件访问地址 /// S3 服务器文件访问地址
/// </summary> /// </summary>
public string ServeUrl { get; set; } public string? ServeUrl { get; set; }
/// <summary> /// <summary>
/// S3服务 Bucket /// S3服务 Bucket
/// </summary> /// </summary>
public string Bucket { get; set; } public string? Bucket { get; set; }
public string AccessKey { get; set; } public string? AccessKey { get; set; }
public string SecretKey { get; set; } public string? SecretKey { get; set; }
} }
+7 -12
View File
@@ -5,20 +5,15 @@ public class SimApiSynapseOptions
/// <summary> /// <summary>
/// Mqtt服务器的Websocket地址 /// Mqtt服务器的Websocket地址
/// </summary> /// </summary>
public string Websocket { get; set; } public string? Websocket { get; set; }
public string? Username { get; set; }
public string Username { get; set; } public string? Password { get; set; }
public string? SysName { get; set; }
public string Password { get; set; } public string? AppName { get; set; }
public string? AppId { get; set; }
public string SysName { get; set; }
public string AppName { get; set; }
public string AppId { get; set; }
public int RpcTimeout { get; set; } = 3; public int RpcTimeout { get; set; } = 3;
public bool EnableConfigStore { get; set; } = true;
public bool DisableEventClient { get; set; } = false; public bool DisableEventClient { get; set; } = false;
public bool DisableRpcClient { get; set; } = false; public bool DisableRpcClient { get; set; } = false;
} }
+2 -2
View File
@@ -18,7 +18,7 @@ public class SimApiBaseController : Controller
/// <summary> /// <summary>
/// 当前登录用户的ID /// 当前登录用户的ID
/// </summary> /// </summary>
protected SimApiLoginItem LoginInfo => (SimApiLoginItem)HttpContext.Items["LoginInfo"]; protected SimApiLoginItem LoginInfo => (SimApiLoginItem)HttpContext.Items["LoginInfo"]!;
/// <summary> /// <summary>
/// 验证请求参数 /// 验证请求参数
@@ -93,7 +93,7 @@ public class SimApiBaseController : Controller
/// <param name="condition">检测条件</param> /// <param name="condition">检测条件</param>
/// <param name="code">错误代码</param> /// <param name="code">错误代码</param>
/// <param name="message">错误描述</param> /// <param name="message">错误描述</param>
protected static void ErrorWhenNull(object condition, int code = 404, string message = "请求的资源不存在") protected static void ErrorWhenNull(object? condition, int code = 404, string message = "请求的资源不存在")
{ {
ErrorWhen(condition == null, code, message); ErrorWhen(condition == null, code, message);
} }
+1 -1
View File
@@ -41,7 +41,7 @@ public class SimApiCommonController(SimApiAuth auth) : SimApiBaseController
[HttpPost("/auth/logout"), SimApiDoc("认证", "退出登陆")] [HttpPost("/auth/logout"), SimApiDoc("认证", "退出登陆")]
public SimApiBaseResponse Logout() public SimApiBaseResponse Logout()
{ {
string token = null; string? token = null;
if (Request.Headers.TryGetValue("Token", out var value)) if (Request.Headers.TryGetValue("Token", out var value))
{ {
+4 -4
View File
@@ -26,10 +26,10 @@ public class SimApiStorage
{ {
var options = apiOptions.SimApiStorageOptions; var options = apiOptions.SimApiStorageOptions;
HttpContextAccessor = httpContextAccessor; HttpContextAccessor = httpContextAccessor;
Endpoint = options.Endpoint; Endpoint = options.Endpoint!;
var useSsl = false; var useSsl = false;
string endpoint; string endpoint;
if (options.Endpoint.StartsWith("http://")) if (options.Endpoint!.StartsWith("http://"))
{ {
endpoint = options.Endpoint.Replace("http://", string.Empty); endpoint = options.Endpoint.Replace("http://", string.Empty);
} }
@@ -43,9 +43,9 @@ public class SimApiStorage
throw new Exception("SimApiStorage: Error Endpoint"); throw new Exception("SimApiStorage: Error Endpoint");
} }
ServeUrl = options.ServeUrl; ServeUrl = options.ServeUrl!;
if (ServeUrl.EndsWith('/')) throw new Exception("SimApiStorage: ServeUrl must not end with /"); if (ServeUrl.EndsWith('/')) throw new Exception("SimApiStorage: ServeUrl must not end with /");
Bucket = options.Bucket; Bucket = options.Bucket!;
var mcb = new MinioClient().WithEndpoint(endpoint) var mcb = new MinioClient().WithEndpoint(endpoint)
.WithCredentials(options.AccessKey, options.SecretKey); .WithCredentials(options.AccessKey, options.SecretKey);
if (useSsl) if (useSsl)
+2 -2
View File
@@ -10,7 +10,7 @@ public class SimApiLogger(string name) : ILogger
public bool IsEnabled(LogLevel logLevel) => true; public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception exception, public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception, string> formatter) Func<TState, Exception, string> formatter)
{ {
Console.ForegroundColor = logLevel switch Console.ForegroundColor = logLevel switch
@@ -23,7 +23,7 @@ public class SimApiLogger(string name) : ILogger
_ => ConsoleColor.White _ => ConsoleColor.White
}; };
var message = var message =
$"[ {name} ][ {DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss:ffff")} ][ {logLevel.ToString()} ]\n{state}\n"; $"[ {name} ][ {DateTime.Now:yyyy-MM-dd HH:mm:ss:ffff} ][ {logLevel.ToString()} ]\n{state}\n";
if (exception != null) if (exception != null)
{ {
message += $"{exception}\n"; message += $"{exception}\n";
-15
View File
@@ -1,15 +0,0 @@
using System;
using System.Collections.Generic;
using Microsoft.Extensions.Logging;
namespace SimApi.Logger;
public class SimApiLoggerConfiguration
{
public int EventId { get; set; }
public Dictionary<LogLevel, ConsoleColor> LogLevels { get; set; } = new()
{
[LogLevel.Information] = ConsoleColor.Green
};
}
+2 -5
View File
@@ -14,21 +14,18 @@ public class SimApiAuthMiddleware(RequestDelegate next)
{ {
public Task Invoke(HttpContext httpContext, IDistributedCache cache, SimApiAuth auth) public Task Invoke(HttpContext httpContext, IDistributedCache cache, SimApiAuth auth)
{ {
string token = null; string? token = null;
if (httpContext.Request.Headers.TryGetValue("Token", out var header)) if (httpContext.Request.Headers.TryGetValue("Token", out var header))
{ {
token = header; token = header;
} }
if (!string.IsNullOrEmpty(token)) if (string.IsNullOrEmpty(token)) return next(httpContext);
{
var login = auth.GetLogin(token); var login = auth.GetLogin(token);
if (login != null) if (login != null)
{ {
httpContext.Items.Add("LoginInfo", login); httpContext.Items.Add("LoginInfo", login);
} }
}
return next(httpContext); return next(httpContext);
} }
} }
+2 -2
View File
@@ -20,7 +20,7 @@ public class SimApiBaseModel
public void MapData<TS>(TS source, bool mapAll = false) public void MapData<TS>(TS source, bool mapAll = false)
{ {
//获取要赋值的源数据不为null的项目 //获取要赋值的源数据不为null的项目
var sourceProps = source.GetType().GetProperties().Where(x => x.GetValue(source) != null) var sourceProps = source!.GetType().GetProperties().Where(x => x.GetValue(source) != null)
.Select(x => new .Select(x => new
{ {
x.Name, x.Name,
@@ -49,7 +49,7 @@ public class SimApiBaseModel
public void MapData<TS>(TS source, string[] mapFields) public void MapData<TS>(TS source, string[] mapFields)
{ {
//获取要赋值的源数据不为null的项目 //获取要赋值的源数据不为null的项目
var sourceProps = source.GetType().GetProperties().Where(x => x.GetValue(source) != null) var sourceProps = source!.GetType().GetProperties().Where(x => x.GetValue(source) != null)
.Select(x => new .Select(x => new
{ {
x.Name, x.Name,
+1
View File
@@ -15,6 +15,7 @@
<GeneratePackageOnBuild>true</GeneratePackageOnBuild> <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
<PackageVersion>5.0.2</PackageVersion> <PackageVersion>5.0.2</PackageVersion>
<TargetFramework>net8.0</TargetFramework> <TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
+56
View File
@@ -0,0 +1,56 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using MQTTnet;
using MQTTnet.Protocol;
namespace SimApi;
public partial class Synapse
{
public event EventHandler<ConfigStoreItem>? OnConfigChanged;
private Dictionary<string, string> CurrentConfig { get; } = new();
private bool FireSetConfig(string key, string value)
{
var topic = $"{Options.SysName}/synapse-config-store/{key}";
var message = new MqttApplicationMessageBuilder()
.WithTopic(topic)
.WithPayload(value)
.WithRetainFlag()
.Build();
if (!Client!.IsConnected) return false;
Client!.PublishAsync(message, CancellationToken.None).Wait();
logger.LogDebug("Synapse Config Set: {Config} => {Data}", key, value);
return true;
}
private string? FireGetConfig(string key)
{
return CurrentConfig.GetValueOrDefault(key);
}
private void RunConfigStoreServer()
{
var csTopicPrefix = $"{Options.SysName}/synapse-config-store/";
var eventSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter(o =>
o.WithTopic($"{csTopicPrefix}+").WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
.Build();
Client!.ApplicationMessageReceivedAsync += e =>
{
if (!e.ApplicationMessage.Topic.StartsWith(csTopicPrefix)) return Task.CompletedTask;
var reqBody = e.ApplicationMessage.ConvertPayloadToString();
var eventName = e.ApplicationMessage.Topic.Replace(csTopicPrefix, string.Empty);
CurrentConfig[eventName] = reqBody;
OnConfigChanged?.Invoke(this, new ConfigStoreItem(eventName, reqBody));
logger.LogDebug("Synapse Config Changed: {Config} => {Data}", eventName, reqBody);
return Task.CompletedTask;
};
Client.SubscribeAsync(eventSubOpts).Wait();
}
public record ConfigStoreItem(string Key, string Value);
}
+14 -6
View File
@@ -8,18 +8,26 @@ namespace SimApi;
public partial class Synapse public partial class Synapse
{ {
private bool FireEvent(string eventName, object param, bool retain = false) private bool FireEvent(string eventName, object? param)
{ {
var paramJson = JsonSerializer.Serialize(param, SimApiUtil.JsonOption); string paramJson;
var topic = $"{Options.SysName}/{Options.AppName}/event/{eventName}"; if (param is string strParam)
{
paramJson = strParam;
}
else
{
paramJson = JsonSerializer.Serialize(param, SimApiUtil.JsonOption);
}
var topic = $"{Options.SysName}/event/{Options.AppName}/{eventName}";
var message = new MqttApplicationMessageBuilder() var message = new MqttApplicationMessageBuilder()
.WithTopic(topic) .WithTopic(topic)
.WithPayload(paramJson) .WithPayload(paramJson)
.WithRetainFlag(retain) .WithRetainFlag(false)
.Build(); .Build();
if (!Client!.IsConnected) return false; if (!Client!.IsConnected) return false;
Client!.PublishAsync(message, CancellationToken.None).Wait(); Client.PublishAsync(message, CancellationToken.None).Wait();
logger.LogDebug("Event Publish: {Event}@{App} {Json}", eventName, Options.AppName, paramJson); logger.LogDebug("Synapse Event Publish: {Event}@{App} {Json}", eventName, Options.AppName, paramJson);
return true; return true;
} }
} }
+32 -18
View File
@@ -1,11 +1,11 @@
using System; using System;
using System.Linq; using System.Linq;
using System.Text.Json; using System.Text.Json;
using System.Text.RegularExpressions;
using System.Threading.Tasks; using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using MQTTnet; using MQTTnet;
using MQTTnet.Protocol;
using SimApi.Helpers; using SimApi.Helpers;
namespace SimApi; namespace SimApi;
@@ -14,36 +14,50 @@ public partial class Synapse
{ {
private void RunEventServer() private void RunEventServer()
{ {
var esTopicPrefix = $"{Options.SysName}/{Options.AppName}/event/"; var esTopicPrefix = $"{Options.SysName}/event/";
var eventSubOpts = MqttFactory.CreateSubscribeOptionsBuilder() Client!.ApplicationMessageReceivedAsync += e =>
.WithTopicFilter(o =>
o.WithTopic($"$queue/{esTopicPrefix}+").WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
.Build();
Client.ApplicationMessageReceivedAsync += e =>
{ {
if (!e.ApplicationMessage.Topic.StartsWith(esTopicPrefix)) return Task.CompletedTask; if (!e.ApplicationMessage.Topic.StartsWith(esTopicPrefix)) return Task.CompletedTask;
var reqBody = e.ApplicationMessage.ConvertPayloadToString(); var reqBody = e.ApplicationMessage.ConvertPayloadToString();
var eventName = e.ApplicationMessage.Topic.Replace(esTopicPrefix, string.Empty); var eventName = e.ApplicationMessage.Topic.Replace(esTopicPrefix, string.Empty);
logger.LogDebug("Synapse Event Receive: {AppName}.{EventName}\n{Body}", logger.LogDebug("Synapse Event Receive: {EventName}\n{Body}", eventName, reqBody);
Options.AppName, eventName, reqBody); var methods = EventRegistry
.Where(x => Regex.IsMatch(eventName,
var method = EventRegistry.FirstOrDefault(x => x.Key == eventName); "^" + Regex.Escape(x.Key!).Replace("\\+", "[^/]+").Replace("\\#", ".*") + "$"))
if (method == null) return Task.CompletedTask; .ToArray();
var callClass = Sp.CreateScope().ServiceProvider.GetRequiredService(method!.Class); foreach (var method in methods)
var mt = callClass.GetType().GetMethod(method.Method); {
var pt = mt!.GetParameters()[0].ParameterType; var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(method.Class!);
var mt = callClass.GetType().GetMethod(method.Method!);
try try
{ {
if (mt!.GetParameters().Length == 2)
{
var pt = mt!.GetParameters()[0].ParameterType;
mt.Invoke(callClass, pt == typeof(string) mt.Invoke(callClass, pt == typeof(string)
? new object[] { reqBody } ? [eventName, reqBody]
: new[] { JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption) }); : [eventName, JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption)]);
}
else
{
mt.Invoke(callClass, [eventName]);
}
} }
catch (Exception ex) catch (Exception ex)
{ {
logger.LogError("Synapse Event Processor Error: {Err}", ex.InnerException); logger.LogError("Synapse Event Processor Error: {Err}", ex.InnerException);
} }
}
return Task.CompletedTask; return Task.CompletedTask;
}; };
Client.SubscribeAsync(eventSubOpts).Wait(); foreach (var ev in EventRegistry)
{
var topic = $"$queue/{esTopicPrefix}{ev.Key}";
var evSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter(o => o.WithTopic(topic)).Build();
Client.SubscribeAsync(evSubOpts).Wait();
logger.LogDebug("Synapse Event Register Event Success: {EventName}\nFull Topic: {Topic}", ev.Key, topic);
}
} }
} }
+13 -4
View File
@@ -19,7 +19,7 @@ public partial class Synapse
var rcTopic = $"{Options.SysName}/{Options.AppName}/rpc/client/{Options.AppId}/"; var rcTopic = $"{Options.SysName}/{Options.AppName}/rpc/client/{Options.AppId}/";
var rcSubOpts = MqttFactory.CreateSubscribeOptionsBuilder() var rcSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter(o => o.WithTopic($"{rcTopic}+")).Build(); .WithTopicFilter(o => o.WithTopic($"{rcTopic}+")).Build();
Client.ApplicationMessageReceivedAsync += e => Client!.ApplicationMessageReceivedAsync += e =>
{ {
if (!e.ApplicationMessage.Topic.StartsWith(rcTopic)) return Task.CompletedTask; if (!e.ApplicationMessage.Topic.StartsWith(rcTopic)) return Task.CompletedTask;
var reqBody = e.ApplicationMessage.ConvertPayloadToString(); var reqBody = e.ApplicationMessage.ConvertPayloadToString();
@@ -32,9 +32,18 @@ public partial class Synapse
Client.SubscribeAsync(rcSubOpts).Wait(); Client.SubscribeAsync(rcSubOpts).Wait();
} }
private string FireRpc(string app, string action, object param) private string? FireRpc(string app, string action, object? param)
{ {
var paramJson = JsonSerializer.Serialize(param, SimApiUtil.JsonOption); string paramJson;
if (param is string strParam)
{
paramJson = strParam;
}
else
{
paramJson = JsonSerializer.Serialize(param, SimApiUtil.JsonOption);
}
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>();
@@ -47,7 +56,7 @@ public partial class Synapse
.WithRetainFlag(false) .WithRetainFlag(false)
.Build(); .Build();
if (!Client!.IsConnected) return null; if (!Client!.IsConnected) return null;
Client!.PublishAsync(message, CancellationToken.None).Wait(); Client.PublishAsync(message, CancellationToken.None).Wait();
logger.LogDebug( logger.LogDebug(
"Synapse RPC Client Request: ({PropsMessageId}) {OptionsAppName} -> {Action}@{App}\n{ParamJson}", messageId, "Synapse RPC Client Request: ({PropsMessageId}) {OptionsAppName} -> {Action}@{App}\n{ParamJson}", messageId,
Options.AppName, action, app, paramJson); Options.AppName, action, app, paramJson);
+20 -9
View File
@@ -23,7 +23,7 @@ public partial class Synapse
.WithTopicFilter(o => .WithTopicFilter(o =>
o.WithTopic($"$queue/{rsTopicPrefix}+").WithRetainHandling(MqttRetainHandling.SendAtSubscribe)) o.WithTopic($"$queue/{rsTopicPrefix}+").WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
.Build(); .Build();
Client.ApplicationMessageReceivedAsync += e => Client!.ApplicationMessageReceivedAsync += e =>
{ {
if (!e.ApplicationMessage.Topic.StartsWith(rsTopicPrefix)) return Task.CompletedTask; if (!e.ApplicationMessage.Topic.StartsWith(rsTopicPrefix)) return Task.CompletedTask;
var reqBody = e.ApplicationMessage.ConvertPayloadToString(); var reqBody = e.ApplicationMessage.ConvertPayloadToString();
@@ -33,19 +33,29 @@ public partial class Synapse
"Synapse RPC Server Receive: ({BasicPropertiesMessageId}) {BasicPropertiesReplyTo} -> {BasicPropertiesType}@{OptionsAppName}\n{S}", "Synapse RPC Server Receive: ({BasicPropertiesMessageId}) {BasicPropertiesReplyTo} -> {BasicPropertiesType}@{OptionsAppName}\n{S}",
e.ApplicationMessage.ContentType, appInfo[0], action, Options.AppName, e.ApplicationMessage.ContentType, appInfo[0], action, Options.AppName,
reqBody); reqBody);
var res = new SimApiBaseResponse(404, "method not found"); 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) return Task.CompletedTask;
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
{ {
var pt = mt!.GetParameters()[0].ParameterType; var methodParams = mt!.GetParameters();
object? ret;
if (methodParams.Length == 0)
{
ret = mt.Invoke(callClass, []);
}
else
{
var pt = mt.GetParameters()[0].ParameterType;
var param = pt == typeof(string) var param = pt == typeof(string)
? [reqBody] ? [reqBody]
: new[] { JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption) }; : new[] { JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption) };
var ret = mt.Invoke(callClass, param); ret = mt.Invoke(callClass, param);
res = new SimApiBaseResponse<object> }
res = new SimApiBaseResponse<object?>
{ {
Data = ret Data = ret
}; };
@@ -67,6 +77,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}";
var message = new MqttApplicationMessageBuilder() var message = new MqttApplicationMessageBuilder()
@@ -74,8 +85,8 @@ public partial class Synapse
.WithPayload(returnJson) .WithPayload(returnJson)
.WithRetainFlag(false) .WithRetainFlag(false)
.Build(); .Build();
if (!Client!.IsConnected) return Task.CompletedTask; if (!Client.IsConnected) return Task.CompletedTask;
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}",
e.ApplicationMessage.ContentType, action, Options.AppName, appInfo[0], returnJson); e.ApplicationMessage.ContentType, action, Options.AppName, appInfo[0], returnJson);
+111 -33
View File
@@ -5,6 +5,7 @@ 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.DependencyInjection;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using MQTTnet; using MQTTnet;
using MQTTnet.Client; using MQTTnet.Client;
@@ -18,20 +19,17 @@ namespace SimApi;
public partial class Synapse(SimApiOptions simApiOptions, ILogger<Synapse> logger, IServiceProvider sp) public partial class Synapse(SimApiOptions simApiOptions, ILogger<Synapse> logger, IServiceProvider sp)
{ {
private IServiceProvider Sp { get; } = sp;
private SimApiSynapseOptions Options { get; } = simApiOptions.SimApiSynapseOptions; private SimApiSynapseOptions Options { get; } = simApiOptions.SimApiSynapseOptions;
private MqttFactory MqttFactory { get; } = new(); private MqttFactory MqttFactory { get; } = new();
public IMqttClient Client { get; set; } public IMqttClient? Client { get; set; }
private List<RegisterItem> EventRegistry { get; set; } private List<RegisterItem> EventRegistry { get; set; } = new();
private List<RegisterItem> RpcRegistry { get; set; } private List<RegisterItem> RpcRegistry { get; set; } = new();
public void Init() public void Init()
{ {
ProcessAttribute();
logger.LogDebug("Synapse初始化配置信息: {Json}", SimApiUtil.Json(Options)); logger.LogDebug("Synapse初始化配置信息: {Json}", SimApiUtil.Json(Options));
if (string.IsNullOrEmpty(Options.AppName) || string.IsNullOrEmpty(Options.SysName)) if (string.IsNullOrEmpty(Options.AppName) || string.IsNullOrEmpty(Options.SysName))
{ {
@@ -42,6 +40,7 @@ public partial class Synapse(SimApiOptions simApiOptions, ILogger<Synapse> logge
logger.LogInformation("Synapse Sys Name: {SysName}\nSynapse App Name: {AppName}\nSynapse App Id: {AppId}", logger.LogInformation("Synapse Sys Name: {SysName}\nSynapse App Name: {AppName}\nSynapse App Id: {AppId}",
Options.SysName, Options.AppName, Options.AppId); Options.SysName, Options.AppName, Options.AppId);
CreateConnection(); CreateConnection();
ProcessAttribute();
//事件客户端 //事件客户端
if (Options.DisableEventClient) if (Options.DisableEventClient)
{ {
@@ -72,10 +71,24 @@ public partial class Synapse(SimApiOptions simApiOptions, ILogger<Synapse> logge
{ {
RunEventServer(); RunEventServer();
} }
if (Options.EnableConfigStore)
{
RunConfigStoreServer();
logger.LogInformation("Synapse Config Store Ready [{SysName}] ...", Options.SysName);
}
} }
public SimApiBaseResponse<T> Rpc<T>(string appName, string method, dynamic param) /// <summary>
/// 调用RPC使用明确的返回值类型
/// </summary>
/// <param name="appName"></param>
/// <param name="method"></param>
/// <param name="param"></param>
/// <typeparam name="T"></typeparam>
/// <returns></returns>
public SimApiBaseResponse<T> Rpc<T>(string appName, string method, dynamic? param = null)
{ {
var res = new SimApiBaseResponse(500, "Synapse Rpc Client Disabled!"); var res = new SimApiBaseResponse(500, "Synapse Rpc Client Disabled!");
if (Options.DisableRpcClient) if (Options.DisableRpcClient)
@@ -88,24 +101,56 @@ public partial class Synapse(SimApiOptions simApiOptions, ILogger<Synapse> logge
res = JsonSerializer.Deserialize<SimApiBaseResponse<T>>(data, SimApiUtil.JsonOption); res = JsonSerializer.Deserialize<SimApiBaseResponse<T>>(data, SimApiUtil.JsonOption);
} }
return res as SimApiBaseResponse<T>; return (res as SimApiBaseResponse<T>)!;
} }
public SimApiBaseResponse<object> Rpc(string appName, string method, dynamic param) /// <summary>
/// 调用RPC使用object作为返回值类型
/// </summary>
/// <param name="appName"></param>
/// <param name="method"></param>
/// <param name="param"></param>
/// <returns></returns>
public SimApiBaseResponse<object> Rpc(string appName, string method, dynamic? param = null)
{ {
return Rpc<object>(appName, method, param); return Rpc<object>(appName, method, param);
} }
public void Event(string eventName, dynamic param) /// <summary>
{ ///
if (Options.DisableEventClient) /// </summary>
/// <param name="eventName"></param>
/// <param name="param"></param>
public bool Event(string eventName, dynamic? param = null)
{ {
if (!Options.DisableEventClient) return FireEvent(eventName, param);
logger.LogError("Synapse Event Client Disabled!"); logger.LogError("Synapse Event Client Disabled!");
return false;
} }
else
/// <summary>
/// 设置一个系统配置项
/// </summary>
/// <param name="key"></param>
/// <param name="value"></param>
/// <returns></returns>
public bool SetConfig(string key, string value)
{ {
FireEvent(eventName, param); if (Options.EnableConfigStore) return FireSetConfig(key, value);
logger.LogError("Synapse Config Store Disabled!");
return false;
} }
/// <summary>
/// 读取一个配置项,如果没有则为空
/// </summary>
/// <param name="key"></param>
/// <returns></returns>
public string? GetConfig(string key)
{
if (Options.EnableConfigStore) return FireGetConfig(key);
logger.LogError("Synapse Config Store Disabled!");
return null;
} }
private void CreateConnection() private void CreateConnection()
@@ -141,8 +186,6 @@ public partial class Synapse(SimApiOptions simApiOptions, ILogger<Synapse> logge
private void ProcessAttribute() private void ProcessAttribute()
{ {
EventRegistry = [];
RpcRegistry = [];
var stackTrace = new StackTrace(); var stackTrace = new StackTrace();
var callingMethod = stackTrace.GetFrame(stackTrace.FrameCount - 1)?.GetMethod(); var callingMethod = stackTrace.GetFrame(stackTrace.FrameCount - 1)?.GetMethod();
var assembly = callingMethod?.DeclaringType?.Assembly; var assembly = callingMethod?.DeclaringType?.Assembly;
@@ -155,49 +198,84 @@ public partial class Synapse(SimApiOptions simApiOptions, ILogger<Synapse> logge
if (method.IsDefined(typeof(SynapseEventAttribute), false)) if (method.IsDefined(typeof(SynapseEventAttribute), false))
{ {
var attribute = var attribute =
(SynapseEventAttribute)Attribute.GetCustomAttribute(method, typeof(SynapseEventAttribute)); (SynapseEventAttribute)Attribute.GetCustomAttribute(method, typeof(SynapseEventAttribute))!;
if (attribute != null) var tmp = new RegisterItem
{
EventRegistry.Add(new RegisterItem
{ {
Key = attribute.Name ?? method.Name, Key = attribute.Name ?? method.Name,
Class = type, Class = type,
Method = method.Name Method = method.Name
}); };
if (tmp.Key.StartsWith('/') || tmp.Key.EndsWith('/'))
{
logger.LogError("Synapse Event Register Error: {Key} Can't start or end of '/'", tmp.Key);
continue;
} }
var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(tmp.Class!);
var mt = callClass.GetType().GetMethod(tmp.Method);
if (mt!.GetParameters().Length > 2 || mt.GetParameters().Length<1)
{
logger.LogError(
"Synapse Event Register Error: Only one or two parameter supported. {Key} -> {Method}@{Class}",
tmp.Key, tmp.Method, tmp.Class.Name);
continue;
}
EventRegistry.Add(tmp);
} }
if (method.IsDefined(typeof(SynapseRpcAttribute), false)) if (method.IsDefined(typeof(SynapseRpcAttribute), false))
{ {
var attribute = var attribute =
(SynapseRpcAttribute)Attribute.GetCustomAttribute(method, typeof(SynapseRpcAttribute)); (SynapseRpcAttribute)Attribute.GetCustomAttribute(method, typeof(SynapseRpcAttribute))!;
if (attribute != null) var tmp = new RegisterItem
{
RpcRegistry.Add(new RegisterItem
{ {
Key = attribute.Name ?? $"{type.Name}.{method.Name}", Key = attribute.Name ?? $"{type.Name}.{method.Name}",
Class = type, Class = type,
Method = method.Name Method = method.Name
}); };
if (tmp.Key.Contains('/') || tmp.Key.Contains('#') || tmp.Key.Contains('+'))
{
logger.LogError("Synapse Rpc Register Error: {Key} contains '/' , '#' , '+'", tmp.Key);
continue;
} }
var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(tmp.Class!);
var mt = callClass.GetType().GetMethod(tmp.Method);
if (mt!.GetParameters().Length > 1)
{
logger.LogError(
"Synapse Rpc Register Error: Only one or none parameter supported. {Key} -> {Method}@{Class}",
tmp.Key, tmp.Method, tmp.Class.Name);
continue;
}
if (RpcRegistry.Any(x => x.Key == tmp.Key))
{
logger.LogError("Synapse Rpc Register Error: {Key} Already Exists -> {Method}@{Class}", tmp.Key,
tmp.Method, tmp.Class.Name);
continue;
}
RpcRegistry.Add(tmp);
} }
} }
} }
var events = EventRegistry.Aggregate(string.Empty, var events = EventRegistry.Aggregate(string.Empty,
(current, ev) => current + $"\n |- {ev.Key} -> {ev.Method}@{ev.Class.Name}"); (current, ev) => current + $"\n |- {ev.Key} -> {ev.Method}@{ev.Class!.Name}");
var rpcList = RpcRegistry.Aggregate(string.Empty, var rpcList = RpcRegistry.Aggregate(string.Empty,
(current, ev) => current + $"\n |- {ev.Key} -> {ev.Method}@{ev.Class.Name}"); (current, ev) => current + $"\n |- {ev.Key} -> {ev.Method}@{ev.Class!.Name}");
if (EventRegistry.Count > 0) logger.LogInformation(">>> Synapse System 读取Event方法:{Event}", events); if (EventRegistry.Count > 0) logger.LogInformation(" >> Synapse System 读取Event方法:{Event}", events);
if (RpcRegistry.Count > 0) logger.LogInformation(">>> Synapse System 读取RPC方法:{Rpc}", rpcList); if (RpcRegistry.Count > 0) logger.LogInformation(" >> Synapse System 读取RPC方法:{Rpc}", rpcList);
} }
} }
public class RegisterItem public class RegisterItem
{ {
public string Key { get; init; } public string? Key { get; init; }
public Type Class { get; init; } public Type? Class { get; init; }
public string Method { get; init; } public string? Method { get; init; }
} }