use lock

Synapse/RpcClient.cs

@@ -43,60 +43,64 @@ public partial class Synapse
     private string? FireRpc(string app, string action, object? param, Dictionary<string, string>? headers = null,
         int? timeout = null)
     {
-        string paramJson;
-        if (param is string strParam)
+        // 检查并发情况下的线程安全问题，使用锁保证字典操作的原子性
+        lock (ResponseCompletionSources)
         {
-            paramJson = strParam;
-        }
-        else
-        {
-            paramJson = JsonSerializer.Serialize(param, SimApiUtil.JsonOption);
-        }
-
-        var topic = $"{Options.SysName}/{app}/rpc/server/{action}";
-        var messageId = Guid.NewGuid().ToString();
-        var tcs = new TaskCompletionSource<string>();
-        ResponseCompletionSources.Add(messageId, tcs);
-        var messageBuilder = new MqttApplicationMessageBuilder()
-            .WithTopic(topic)
-            .WithPayload(paramJson)
-            .WithResponseTopic($"{Options.AppName},{Options.AppId},{messageId}")
-            .WithContentType("application/json")
-            .WithRetainFlag(false);
-        foreach (var h in headers ?? new Dictionary<string, string>())
-        {
-            messageBuilder.WithUserProperty(h.Key, h.Value);
-        }
-
-        var message = messageBuilder.Build();
-        if (!Client!.IsConnected) return null;
-        Client.PublishAsync(message, CancellationToken.None).Wait();
-        logger.LogDebug(
-            "Synapse RPC Client Request: ({PropsMessageId}) {OptionsAppName} -> {Action}@{App}\n{ParamJson}\nHeaders: {Headers}",
-            messageId, Options.AppName, action, app, paramJson, JsonSerializer.Serialize(headers));
-
-        string response;
-        try
-        {
-            timeout ??= Options.RpcTimeout;
-            if (tcs.Task.Wait(timeout.Value * 1000))
+            string paramJson;
+            if (param is string strParam)
             {
-                response = tcs.Task.Result;
-                logger.LogDebug(
-                    "Synapse RPC Client Response: ({BasicPropertiesCorrelationId}) {BasicPropertiesType}@{BasicPropertiesReplyTo} -> {OptionsAppName}\n{S}",
-                    messageId, action, app, Options.AppName, response);
+                paramJson = strParam;
             }
             else
             {
-                response = JsonSerializer.Serialize(new SimApiBaseResponse(502, "timeout"), SimApiUtil.JsonOption);
+                paramJson = JsonSerializer.Serialize(param, SimApiUtil.JsonOption);
             }
-        }
-        catch
-        {
-            response = JsonSerializer.Serialize(new SimApiBaseResponse(500, "Synapse RPC Client Error"),
-                SimApiUtil.JsonOption);
-        }
 
-        return response;
+            var topic = $"{Options.SysName}/{app}/rpc/server/{action}";
+            var messageId = Guid.NewGuid().ToString();
+            var tcs = new TaskCompletionSource<string>();
+            ResponseCompletionSources.Add(messageId, tcs);
+            var messageBuilder = new MqttApplicationMessageBuilder()
+                .WithTopic(topic)
+                .WithPayload(paramJson)
+                .WithResponseTopic($"{Options.AppName},{Options.AppId},{messageId}")
+                .WithContentType("application/json")
+                .WithRetainFlag(false);
+            foreach (var h in headers ?? new Dictionary<string, string>())
+            {
+                messageBuilder.WithUserProperty(h.Key, h.Value);
+            }
+
+            var message = messageBuilder.Build();
+            if (!Client!.IsConnected) return null;
+            Client.PublishAsync(message, CancellationToken.None).Wait();
+            logger.LogDebug(
+                "Synapse RPC Client Request: ({PropsMessageId}) {OptionsAppName} -> {Action}@{App}\n{ParamJson}\nHeaders: {Headers}",
+                messageId, Options.AppName, action, app, paramJson, JsonSerializer.Serialize(headers));
+
+            string response;
+            try
+            {
+                timeout ??= Options.RpcTimeout;
+                if (tcs.Task.Wait(timeout.Value * 1000))
+                {
+                    response = tcs.Task.Result;
+                    logger.LogDebug(
+                        "Synapse RPC Client Response: ({BasicPropertiesCorrelationId}) {BasicPropertiesType}@{BasicPropertiesReplyTo} -> {OptionsAppName}\n{S}",
+                        messageId, action, app, Options.AppName, response);
+                }
+                else
+                {
+                    response = JsonSerializer.Serialize(new SimApiBaseResponse(502, "timeout"), SimApiUtil.JsonOption);
+                }
+            }
+            catch
+            {
+                response = JsonSerializer.Serialize(new SimApiBaseResponse(500, "Synapse RPC Client Error"),
+                    SimApiUtil.JsonOption);
+            }
+
+            return response;
+        }
     }
 }

Synapse/RpcServer.cs

@@ -20,9 +20,9 @@ public partial class Synapse
 
     private void RunRpcServer()
     {
-        Client!.ApplicationMessageReceivedAsync += e =>
+        Client!.ApplicationMessageReceivedAsync += async e =>
         {
-            Task.Run(() =>
+            await Task.Run(() =>
             {
                 if (!e.ApplicationMessage.Topic.StartsWith(RpcServerTopicPrefix)) return;
                 var reqBody = e.ApplicationMessage.ConvertPayloadToString();
@@ -109,7 +109,6 @@ public partial class Synapse
                     "Synapse Rpc Server Return: ({BasicPropertiesMessageId}) {BasicPropertiesType}@{OptionsAppName} -> {BasicPropertiesReplyTo}\n{ReturnJson}",
                     appInfo[2], action, Options.AppName, appInfo[0], returnJson);
             });
-            return Task.CompletedTask;
         };
         SubRpcServerTopic();
     }