using System; using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Text.Json; using Microsoft.Extensions.Logging; using RabbitMQ.Client; using RabbitMQ.Client.Exceptions; using SimApi.Attributes; using SimApi.Communications; using SimApi.Configs; using SimApi.Helpers; namespace SimApi; public partial class Synapse { private IServiceProvider Sp { get; } private SimApiSynapseOptions Options { get; } private ILogger Logger { get; } private IConnection Connection { get; set; } private IModel EventClientChannel { get; set; } private IModel EventServerChannel { get; set; } private IModel RpcClientChannel { get; set; } private IModel RpcServerChannel { get; set; } private List EventRegistry { get; set; } private List RpcRegistry { get; set; } public Synapse(SimApiOptions simApiOptions, ILogger logger, IServiceProvider sp) { Logger = logger; Sp = sp; Options = simApiOptions.SimApiSynapseOptions; } public void Init() { ProcessAttribute(); Logger.LogDebug("Synapse初始化配置信息: {Json}", SimApiUtil.Json(Options)); if (string.IsNullOrEmpty(Options.AppName) || string.IsNullOrEmpty(Options.SysName)) { Logger.LogCritical("Synapse初始化失败: AppName or SysName 错误"); } Options.AppId ??= Guid.NewGuid().ToString(); Logger.LogInformation("System Name: {SysName}\nApp Name: {AppName}\nAppId: {AppId}", Options.SysName, Options.AppName, Options.AppId); CreateConnection(); CheckAndCreateExchange(); //事件客户端 if (Options.DisableEventClient) { Logger.LogWarning("Event Client Disabled: DisableEventClient set true"); } else { RunEventClient(); Logger.LogInformation("Event Client Ready"); } //RPC客户端 if (Options.DisableRpcClient) { Logger.LogWarning("Rpc Client Disabled: DisableEventClient set true"); } else { RunRpcClient(); Logger.LogInformation("Rpc Client Ready, Client Timeout: {OptionsRpcTimeout}s", Options.RpcTimeout); } if (RpcRegistry.Count > 0) { RunRpcServer(); } if (EventRegistry.Count > 0) { RunEventServer(); } } public SimApiBaseResponse Rpc(string appName, string method, dynamic param) { var res = new SimApiBaseResponse(500, "Rpc Client Disabled!"); if (Options.DisableRpcClient) { Logger.LogError("Rpc Client Disabled!"); } else { var data = FireRpc(appName, method, param); res = JsonSerializer.Deserialize>(data, SimApiUtil.JsonOption); } return res as SimApiBaseResponse; } public SimApiBaseResponse Rpc(string appName, string method, dynamic param) { return Rpc(appName, method, param); } public void Event(string eventName, dynamic param) { if (Options.DisableEventClient) { Logger.LogError("Event Client Disabled!"); } else { FireEvent(eventName, param); } } private void CreateConnection() { var factory = new ConnectionFactory { HostName = Options.MqHost, Port = Options.MqPort, VirtualHost = Options.MqVHost, UserName = Options.MqUser, Password = Options.MqPass }; try { Connection = factory.CreateConnection(); Logger.LogInformation("连接RabbitMQ服务器成功"); } catch (BrokerUnreachableException e) { Logger.LogError("连接RabbitMQ失败: \n{Err}", e); } } private IModel CreateChannel(ushort processNum = 0, string desc = "unknow") { IModel channel = null; try { var log = $"Channel [{desc}] 创建成功..."; channel = Connection.CreateModel(); if (processNum != 0) { channel.BasicQos(0, processNum, false); log += $"最大处理器数量: {processNum}"; } Logger.LogInformation(log); } catch (ConnectFailureException e) { Logger.LogError("Channel [{{Desc}}] 创建失败...\n {0}", e); } return channel; } private void CheckAndCreateExchange() { var channel = CreateChannel(0, "Exchange"); try { channel.ExchangeDeclare(Options.SysName, ExchangeType.Topic, true, true, null); Logger.LogDebug("Register Exchange Success"); } catch (ConnectFailureException e) { Logger.LogError("Failed to declare Exchange.\n {Err}", e); } channel.Close(); Logger.LogDebug("Exchange Channel Closed"); } private void ProcessAttribute() { EventRegistry = new List(); RpcRegistry = new List(); var stackTrace = new StackTrace(); var callingMethod = stackTrace.GetFrame(stackTrace.FrameCount - 1).GetMethod(); var assembly = callingMethod.DeclaringType.Assembly; var types = assembly.GetTypes(); // 获取程序集中的所有类型 foreach (var type in types) { var methods = type.GetMethods(); // 获取类型中的所有方法 foreach (var method in methods) { if (method.IsDefined(typeof(SynapseEventAttribute), false)) { var attribute = (SynapseEventAttribute)Attribute.GetCustomAttribute(method, typeof(SynapseEventAttribute)); if (attribute != null) { EventRegistry.Add(new RegisterItem { Key = attribute.Name ?? method.Name, Class = type, Method = method.Name }); } } if (method.IsDefined(typeof(SynapseRpcAttribute), false)) { var attribute = (SynapseRpcAttribute)Attribute.GetCustomAttribute(method, typeof(SynapseRpcAttribute)); if (attribute != null) { RpcRegistry.Add(new RegisterItem { Key = attribute.Name ?? $"{type.Name}.{method.Name}", Class = type, Method = method.Name }); } } } } var events = EventRegistry.Aggregate(string.Empty, (current, ev) => current + $"\n |- {ev.Key} -> {ev.Method}@{ev.Class.Name}"); var rpcList = RpcRegistry.Aggregate(string.Empty, (current, ev) => current + $"\n |- {ev.Key} -> {ev.Method}@{ev.Class.Name}"); Logger.LogInformation(">>> Synapse System 读取Event方法:{Event}\n>>>Synapse System 读取RPC方法:{Rpc}", events, rpcList); } } public class RegisterItem { public string Key { get; set; } public Type Class { get; set; } public string Method { get; set; } }