add event loadbalancing

Configurations/SimApiSynapseOptions.cs

@@ -13,6 +13,12 @@ public class SimApiSynapseOptions
     public string? AppId { get; set; }
     public int RpcTimeout { get; set; } = 3;
 
+    /// <summary>
+    /// Event是否使用负载均衡
+    /// 也就是订阅$queue主题,消息会分发给不同的AppId
+    /// 如果false,多个AppId都可以同时收到消息
+    /// </summary>
+    public bool EventLoadBalancing { get; set; } = false;
     public bool EnableConfigStore { get; set; } = true;
     public bool DisableEventClient { get; set; } = false;
     public bool DisableRpcClient { get; set; } = false;

SimApiExtensions.cs

@@ -224,11 +224,11 @@ public static class SimApiExtensions
     /// </summary>
     /// <param name="builder"></param>
     /// <returns></returns>
-    public static IApplicationBuilder UseSimApi(this IApplicationBuilder builder)
+    public static WebApplication UseSimApi(this WebApplication builder)
     {
-        var options = builder.ApplicationServices.GetRequiredService<SimApiOptions>();
+        var options = builder.Services.GetRequiredService<SimApiOptions>();
 
-        var logger = builder.ApplicationServices.GetRequiredService<ILogger<SimApiOptions>>();
+        var logger = builder.Services.GetRequiredService<ILogger<SimApiOptions>>();
 
         logger.LogInformation("当前时区: {LocalId}", TimeZoneInfo.Local.Id);
         if (options.EnableForwardHeaders)
@@ -277,7 +277,7 @@ public static class SimApiExtensions
         if (options.EnableSimApiStorage)
         {
             logger.LogInformation("开始配置SimApiStorage...");
-            builder.ApplicationServices.GetService<SimApiStorage>();
+            builder.Services.GetService<SimApiStorage>();
         }
 
         if (options.EnableLowerUrl)
@@ -287,7 +287,7 @@ public static class SimApiExtensions
 
         if (options.EnableSynapse)
         {
-            var synapse = builder.ApplicationServices.GetRequiredService<Synapse>();
+            var synapse = builder.Services.GetRequiredService<Synapse>();
             synapse.Init();
         }
 

Synapse/ConfigStore.cs

@@ -15,6 +15,7 @@ public partial class Synapse
 
     private bool FireSetConfig(string key, string value)
     {
+        if (key.Contains('#') || key.Contains('+')) return false;
         var topic = $"{Options.SysName}/synapse-config-store/{key}";
         var message = new MqttApplicationMessageBuilder()
             .WithTopic(topic)
@@ -37,7 +38,7 @@ public partial class Synapse
         var csTopicPrefix = $"{Options.SysName}/synapse-config-store/";
         var eventSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
             .WithTopicFilter(o =>
-                o.WithTopic($"{csTopicPrefix}+").WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
+                o.WithTopic($"{csTopicPrefix}#").WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
             .Build();
         Client!.ApplicationMessageReceivedAsync += e =>
         {

Synapse/EventServer.cs

@@ -33,7 +33,7 @@ public partial class Synapse
                 {
                     if (mt!.GetParameters().Length == 2)
                     {
-                        var pt = mt!.GetParameters()[0].ParameterType;
+                        var pt = mt.GetParameters()[1].ParameterType;
                         mt.Invoke(callClass, pt == typeof(string)
                             ? [eventName, reqBody]
                             : [eventName, JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption)]);
@@ -45,7 +45,7 @@ public partial class Synapse
                 }
                 catch (Exception ex)
                 {
-                    logger.LogError("Synapse Event Processor Error: {Err}", ex.InnerException);
+                    logger.LogError("Synapse Event Processor Error: {Err}\n{Stack}", ex.Message,ex.StackTrace);
                 }
             }
 
@@ -53,7 +53,12 @@ public partial class Synapse
         };
         foreach (var ev in EventRegistry)
         {
-            var topic = $"$queue/{esTopicPrefix}{ev.Key}";
+            
+            var topic = $"{esTopicPrefix}{ev.Key}";
+            if (Options.EventLoadBalancing)
+            {
+                topic = "$queue/" + topic;
+            }
             var evSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
                 .WithTopicFilter(o => o.WithTopic(topic)).Build();
             Client.SubscribeAsync(evSubOpts).Wait();