102 lines
3.7 KiB
C#
102 lines
3.7 KiB
C#
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;
|
|
|
|
namespace SimApi;
|
|
|
|
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()
|
|
{
|
|
Client!.ApplicationMessageReceivedAsync += e =>
|
|
{
|
|
if (!e.ApplicationMessage.Topic.StartsWith(EventClientTopicPrefix)) return Task.CompletedTask;
|
|
var reqBody = e.ApplicationMessage.ConvertPayloadToString();
|
|
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;
|
|
};
|
|
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, Dictionary<string, string>? headers = null,
|
|
int? timeout = null)
|
|
{
|
|
string paramJson;
|
|
if (param is string strParam)
|
|
{
|
|
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))
|
|
{
|
|
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;
|
|
}
|
|
} |