fix warning

2024-08-03 19:19:42 +08:00
parent 4d0f801f21
commit ac4619a789
10 changed files with 122 additions and 86 deletions
@@ -13,6 +13,8 @@ public partial class Synapse
    public event EventHandler<ConfigStoreItem>? OnConfigChanged;
    private Dictionary<string, string> CurrentConfig { get; } = new();

+    private string ConfigStoreTopicPrefix => $"{Options.SysName}/synapse-config-store/";
+
    private bool FireSetConfig(string key, string value)
    {
        if (key.Contains('#') || key.Contains('+')) return false;
@@ -35,22 +37,26 @@ public partial class Synapse

    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;
+            if (!e.ApplicationMessage.Topic.StartsWith(ConfigStoreTopicPrefix)) return Task.CompletedTask;
            var reqBody = e.ApplicationMessage.ConvertPayloadToString();
-            var eventName = e.ApplicationMessage.Topic.Replace(csTopicPrefix, string.Empty);
+            var eventName = e.ApplicationMessage.Topic.Replace(ConfigStoreTopicPrefix, 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();
+        SubConfigStoreServerTopic();
+    }
+
+    private void SubConfigStoreServerTopic()
+    {
+        var csSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
+            .WithTopicFilter(o =>
+                o.WithTopic($"{ConfigStoreTopicPrefix}#").WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
+            .Build();
+        Client!.SubscribeAsync(csSubOpts).Wait();
    }

    public record ConfigStoreItem(string Key, string Value);
@@ -12,14 +12,16 @@ namespace SimApi;

 public partial class Synapse
 {
+    
+    private string EventServerTopicPrefix => $"{Options.SysName}/event/";
+
    private void RunEventServer()
    {
-        var esTopicPrefix = $"{Options.SysName}/event/";
        Client!.ApplicationMessageReceivedAsync += e =>
        {
-            if (!e.ApplicationMessage.Topic.StartsWith(esTopicPrefix)) return Task.CompletedTask;
+            if (!e.ApplicationMessage.Topic.StartsWith(EventServerTopicPrefix)) return Task.CompletedTask;
            var reqBody = e.ApplicationMessage.ConvertPayloadToString();
-            var eventName = e.ApplicationMessage.Topic.Replace(esTopicPrefix, string.Empty);
+            var eventName = e.ApplicationMessage.Topic.Replace(EventServerTopicPrefix, string.Empty);
            logger.LogDebug("Synapse Event Receive: {EventName}\n{Body}", eventName, reqBody);
            var methods = EventRegistry
                .Where(x => Regex.IsMatch(eventName,
@@ -48,20 +50,23 @@ public partial class Synapse
                    logger.LogError("Synapse Event Processor Error: {Err}\n{Stack}", ex.Message,ex.StackTrace);
                }
            }
-
            return Task.CompletedTask;
        };
+        SubRpcServerTopic();
+    }
+
+    private void SubEventServerTopic()
+    {
        foreach (var ev in EventRegistry)
        {
-            
-            var topic = $"{esTopicPrefix}{ev.Key}";
+            var topic = $"{EventServerTopicPrefix}{ev.Key}";
            if (Options.EventLoadBalancing)
            {
                topic = "$queue/" + topic;
            }
            var evSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
                .WithTopicFilter(o => o.WithTopic(topic)).Build();
-            Client.SubscribeAsync(evSubOpts).Wait();
+            Client!.SubscribeAsync(evSubOpts).Wait();
            logger.LogDebug("Synapse Event Register Event Success: {EventName}\nFull Topic: {Topic}", ev.Key, topic);
        }
    }
@@ -14,22 +14,28 @@ public partial class Synapse
 {
    private Dictionary<string, TaskCompletionSource<string>> ResponseCompletionSources { get; } = new();

+    private string EventClientTopicPrefix => $"{Options.SysName}/{Options.AppName}/rpc/client/{Options.AppId}/";
+
    private void RunRpcClient()
    {
-        var rcTopic = $"{Options.SysName}/{Options.AppName}/rpc/client/{Options.AppId}/";
-        var rcSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
-            .WithTopicFilter(o => o.WithTopic($"{rcTopic}+")).Build();
        Client!.ApplicationMessageReceivedAsync += e =>
        {
-            if (!e.ApplicationMessage.Topic.StartsWith(rcTopic)) return Task.CompletedTask;
+            if (!e.ApplicationMessage.Topic.StartsWith(EventClientTopicPrefix)) return Task.CompletedTask;
            var reqBody = e.ApplicationMessage.ConvertPayloadToString();
-            var messageId = e.ApplicationMessage.Topic.Replace(rcTopic, string.Empty);
+            var messageId = e.ApplicationMessage.Topic.Replace(EventClientTopicPrefix, string.Empty);
            if (!ResponseCompletionSources.TryGetValue(messageId, out var tcs)) return Task.CompletedTask;
            tcs.SetResult(reqBody);
            ResponseCompletionSources.Remove(messageId);
            return Task.CompletedTask;
        };
-        Client.SubscribeAsync(rcSubOpts).Wait();
+        SubRpcClientTopic();
+    }
+
+    private void SubRpcClientTopic()
+    {
+        var rcSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
+            .WithTopicFilter(o => o.WithTopic($"{EventClientTopicPrefix}+")).Build();
+        Client!.SubscribeAsync(rcSubOpts).Wait();
    }

    private string? FireRpc(string app, string action, object? param)
@@ -16,18 +16,15 @@ namespace SimApi;

 public partial class Synapse
 {
+    private string RpcServerTopicPrefix => $"{Options.SysName}/{Options.AppName}/rpc/server/";
+
    private void RunRpcServer()
    {
-        var rsTopicPrefix = $"{Options.SysName}/{Options.AppName}/rpc/server/";
-        var rsSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
-            .WithTopicFilter(o =>
-                o.WithTopic($"$queue/{rsTopicPrefix}+").WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
-            .Build();
        Client!.ApplicationMessageReceivedAsync += e =>
        {
-            if (!e.ApplicationMessage.Topic.StartsWith(rsTopicPrefix)) return Task.CompletedTask;
+            if (!e.ApplicationMessage.Topic.StartsWith(RpcServerTopicPrefix)) return Task.CompletedTask;
            var reqBody = e.ApplicationMessage.ConvertPayloadToString();
-            var action = e.ApplicationMessage.Topic.Replace(rsTopicPrefix, string.Empty);
+            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}",
@@ -93,6 +90,15 @@ public partial class Synapse

            return Task.CompletedTask;
        };
-        Client.SubscribeAsync(rsSubOpts).Wait();
+        SubRpcServerTopic();
+    }
+
+    public void SubRpcServerTopic()
+    {
+        var rsSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
+            .WithTopicFilter(o =>
+                o.WithTopic($"$queue/{RpcServerTopicPrefix}+").WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
+            .Build();
+        Client!.SubscribeAsync(rsSubOpts).Wait();
    }
 }
@@ -41,8 +41,45 @@ public partial class Synapse(SimApiOptions simApiOptions, ILogger<Synapse> logge
            Options.SysName, Options.AppName, Options.AppId);
        CreateConnection();
        ProcessAttribute();
+        //事件客户端
+        if (Options.DisableEventClient)
+        {
+            logger.LogWarning("Synapse Event Client Disabled: DisableEventClient set true");
+        }
+        else
+        {
+            logger.LogInformation("Synapse Event Client Ready");
+        }
+
+        //RPC客户端
+        if (Options.DisableRpcClient)
+        {
+            logger.LogWarning("Synapse Rpc Client Disabled: DisableEventClient set true");
+        }
+        else
+        {
+            RunRpcClient();
+            logger.LogInformation("Synapse Rpc Client Ready, Client Timeout: {OptionsRpcTimeout}s", Options.RpcTimeout);
+        }
+
+        if (RpcRegistry.Count > 0)
+        {
+            RunRpcServer();
+        }
+
+        if (EventRegistry.Count > 0)
+        {
+            RunEventServer();
+        }
+
+        if (Options.EnableConfigStore)
+        {
+            RunConfigStoreServer();
+            logger.LogInformation("Synapse Config Store Ready [{SysName}] ...", Options.SysName);
+        }
    }
-    
+
+
    /// <summary>
    /// 调用RPC使用明确的返回值类型
    /// </summary>
@@ -128,49 +165,18 @@ public partial class Synapse(SimApiOptions simApiOptions, ILogger<Synapse> logge
        Client.ConnectedAsync += _ =>
        {
            logger.LogInformation("Synapse MQTT[{AppName}:{AppId}] 连接成功...", Options.AppName, Options.AppId);
-            //事件客户端
-            if (Options.DisableEventClient)
-            {
-                logger.LogWarning("Synapse Event Client Disabled: DisableEventClient set true");
-            }
-            else
-            {
-                logger.LogInformation("Synapse Event Client Ready");
-            }
-
            //RPC客户端
-            if (Options.DisableRpcClient)
-            {
-                logger.LogWarning("Synapse Rpc Client Disabled: DisableEventClient set true");
-            }
-            else
-            {
-                RunRpcClient();
-                logger.LogInformation("Synapse Rpc Client Ready, Client Timeout: {OptionsRpcTimeout}s", Options.RpcTimeout);
-            }
-
-            if (RpcRegistry.Count > 0)
-            {
-                RunRpcServer();
-            }
-
-            if (EventRegistry.Count > 0)
-            {
-                RunEventServer();
-            }
-
-            if (Options.EnableConfigStore)
-            {
-                RunConfigStoreServer();
-                logger.LogInformation("Synapse Config Store Ready [{SysName}] ...", Options.SysName);
-            }
+            if (!Options.DisableRpcClient) SubRpcClientTopic();
+            if (RpcRegistry.Count > 0) SubRpcServerTopic();
+            if (EventRegistry.Count > 0) SubEventServerTopic();
+            if (Options.EnableConfigStore) SubConfigStoreServerTopic();
            return Task.CompletedTask;
        };
        //重连
        Client.DisconnectedAsync += async _ =>
        {
            logger.LogError("Synapse MQTT[{AppName}:{AppId}] 断开连接,开始重连...", Options.AppName, Options.AppId);
-            await Task.Delay(TimeSpan.FromSeconds(3));
+            await Task.Delay(TimeSpan.FromSeconds(5));
            try
            {
                logger.LogInformation("Synapse MQTT[{AppName}:{AppId}] 开始连接MQTT服务器...", Options.AppName, Options.AppId);