Compare commits

...
6 Commits
Author SHA1 Message Date
xrain 4e1b7d4845 update to 9 2025-10-14 02:20:37 +08:00
xrain 71acb35df9 cancle json refrence 2025-08-13 17:28:24 +08:00
xrain 67cc7346dc use concurrentDic 2025-05-17 21:17:43 +08:00
xrain 8627ae61fa use lock 2025-05-17 21:11:27 +08:00
xrain 02523028db add task.run 2025-05-17 20:40:48 +08:00
xrain 7d39ab5e0b add task.run 2025-05-17 20:38:39 +08:00
6 changed files with 116 additions and 111 deletions
+2 -2
View File
@@ -24,10 +24,10 @@ public static class SimApiUtil
/// </summary>
public static JsonSerializerOptions JsonOption => new()
{
ReferenceHandler = ReferenceHandler.Preserve,
// ReferenceHandler = ReferenceHandler.Preserve,
Encoder = JavaScriptEncoder.Create(UnicodeRanges.All),
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull
// DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull
};
/// <summary>
+1 -1
View File
@@ -14,8 +14,8 @@
<SynchReleaseVersion>false</SynchReleaseVersion>
<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
<PackageVersion>5.0.2</PackageVersion>
<TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable>
<TargetFrameworks>net8.0;net9.0</TargetFrameworks>
</PropertyGroup>
<ItemGroup>
-2
View File
@@ -3,7 +3,6 @@ using System.Diagnostics;
using System.Linq;
using System.Reflection;
using System.Text.Json;
using System.Text.Json.Serialization;
using Hangfire;
using Hangfire.Console;
using Hangfire.Redis.StackExchange;
@@ -251,7 +250,6 @@ public static class SimApiExtensions
.AddJsonOptions(opt =>
{
opt.JsonSerializerOptions.PropertyNamingPolicy = JsonNamingPolicy.CamelCase;
opt.JsonSerializerOptions.ReferenceHandler = ReferenceHandler.Preserve;
});
}
+5 -2
View File
@@ -18,6 +18,8 @@ public partial class Synapse
private void RunEventServer()
{
Client!.ApplicationMessageReceivedAsync += async e =>
{
await Task.Run(async () =>
{
if (!e.ApplicationMessage.Topic.StartsWith(EventServerTopicPrefix)) return;
var reqBody = e.ApplicationMessage.ConvertPayloadToString();
@@ -37,8 +39,8 @@ public partial class Synapse
{
var pt = mt.GetParameters()[1].ParameterType;
mt.Invoke(callClass, pt == typeof(string)
? new object?[] { eventName, reqBody }
: new object?[] { eventName, JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption) });
? [eventName, reqBody]
: [eventName, JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption)]);
}
else
{
@@ -52,6 +54,7 @@ public partial class Synapse
}))
.ToList();
await Task.WhenAll(tasks);
});
};
SubEventServerTopic();
}
+5 -5
View File
@@ -1,20 +1,19 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using MQTTnet;
using MQTTnet.Packets;
using SimApi.Communications;
using SimApi.Helpers;
using System.Collections.Concurrent;
namespace SimApi;
public partial class Synapse
{
private Dictionary<string, TaskCompletionSource<string>> ResponseCompletionSources { get; } = new();
private ConcurrentDictionary<string, TaskCompletionSource<string>> ResponseCompletionSources { get; } = new();
private string EventClientTopicPrefix => $"{Options.SysName}/{Options.AppName}/rpc/client/{Options.AppId}/";
@@ -27,7 +26,7 @@ public partial class Synapse
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);
ResponseCompletionSources.Remove(messageId, out _);
return Task.CompletedTask;
};
SubRpcClientTopic();
@@ -43,6 +42,7 @@ 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)
{
@@ -56,7 +56,7 @@ public partial class Synapse
var topic = $"{Options.SysName}/{app}/rpc/server/{action}";
var messageId = Guid.NewGuid().ToString();
var tcs = new TaskCompletionSource<string>();
ResponseCompletionSources.Add(messageId, tcs);
ResponseCompletionSources.TryAdd(messageId, tcs);
var messageBuilder = new MqttApplicationMessageBuilder()
.WithTopic(topic)
.WithPayload(paramJson)
+10 -6
View File
@@ -20,9 +20,11 @@ public partial class Synapse
private void RunRpcServer()
{
Client!.ApplicationMessageReceivedAsync += e =>
Client!.ApplicationMessageReceivedAsync += async e =>
{
if (!e.ApplicationMessage.Topic.StartsWith(RpcServerTopicPrefix)) return Task.CompletedTask;
await Task.Run(() =>
{
if (!e.ApplicationMessage.Topic.StartsWith(RpcServerTopicPrefix)) return;
var reqBody = e.ApplicationMessage.ConvertPayloadToString();
var action = e.ApplicationMessage.Topic.Replace(RpcServerTopicPrefix, string.Empty);
var appInfo = e.ApplicationMessage.ResponseTopic.Split(",");
@@ -58,7 +60,10 @@ public partial class Synapse
var pt2 = mt.GetParameters()[0].ParameterType;
var param2 = pt2 == typeof(string)
? [reqBody]
: new[] { JsonSerializer.Deserialize(reqBody, pt2, SimApiUtil.JsonOption), headerData };
: new[]
{
JsonSerializer.Deserialize(reqBody, pt2, SimApiUtil.JsonOption), headerData
};
ret = mt.Invoke(callClass, param2);
break;
default:
@@ -98,13 +103,12 @@ public partial class Synapse
.WithPayload(returnJson)
.WithRetainFlag(false)
.Build();
if (!Client.IsConnected) return Task.CompletedTask;
if (!Client.IsConnected) return;
Client.PublishAsync(message, CancellationToken.None).Wait();
logger.LogDebug(
"Synapse Rpc Server Return: ({BasicPropertiesMessageId}) {BasicPropertiesType}@{OptionsAppName} -> {BasicPropertiesReplyTo}\n{ReturnJson}",
appInfo[2], action, Options.AppName, appInfo[0], returnJson);
return Task.CompletedTask;
});
};
SubRpcServerTopic();
}