96 lines
2.8 KiB
C#
96 lines
2.8 KiB
C#
using MasstransferCommon.Utils;
|
|
using MasstransferInfrastructure.Mqtt.Model;
|
|
using MQTTnet;
|
|
using MQTTnet.Client;
|
|
using MQTTnet.Protocol;
|
|
using Serilog;
|
|
|
|
namespace MasstransferInfrastructure.Mqtt.Client;
|
|
|
|
public class MessageQueueHelper<T>
|
|
{
|
|
private static readonly Dictionary<string, List<Action<string, T>>> Subscribers = new();
|
|
|
|
// ReSharper disable once StaticMemberInGenericType
|
|
private static readonly MqttClient Client = new();
|
|
|
|
|
|
/// <summary>
|
|
/// 初始化连接
|
|
/// </summary>
|
|
/// <param name="options"></param>
|
|
public static async Task<bool> InitConnect(MqttConnectOptions options)
|
|
{
|
|
try
|
|
{
|
|
if (!await Client.ConnectAsync(options)) return false;
|
|
// 连接成功后监听消息
|
|
Client.MessageReceived += HandleMessageReceived;
|
|
return true;
|
|
}
|
|
catch (Exception e)
|
|
{
|
|
Log.Error(e, "连接MQTT服务器失败");
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 订阅某个主题
|
|
/// </summary>
|
|
/// <param name="topic"></param>
|
|
/// <param name="delegate"></param>
|
|
/// <param name="qos"></param>
|
|
public static async Task<bool> Subscribe(string topic, Action<string, T> @delegate,
|
|
MqttQualityOfServiceLevel qos = MqttQualityOfServiceLevel.AtMostOnce)
|
|
{
|
|
if (!Subscribers.ContainsKey(topic))
|
|
{
|
|
Subscribers.Add(topic, []);
|
|
}
|
|
|
|
Subscribers[topic].Add(@delegate);
|
|
|
|
return await Client.Subscribe(topic, qos);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 发送消息
|
|
/// </summary>
|
|
/// <param name="topic"></param>
|
|
/// <param name="message"></param>
|
|
/// <param name="qos"></param>
|
|
public static async Task<bool> Publish(string topic, object message,
|
|
MqttQualityOfServiceLevel qos = MqttQualityOfServiceLevel.AtMostOnce)
|
|
{
|
|
return await Client.Publish(topic, message, qos);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 处理接收到的消息
|
|
/// </summary>
|
|
/// <param name="sender"></param>
|
|
/// <param name="e"></param>
|
|
/// <returns></returns>
|
|
private static void HandleMessageReceived(object? sender, MqttApplicationMessageReceivedEventArgs e)
|
|
{
|
|
var applicationMessage = e.ApplicationMessage;
|
|
var topic = applicationMessage.Topic;
|
|
var message = applicationMessage.ConvertPayloadToString();
|
|
|
|
if (!Subscribers.TryGetValue(topic, out var subscribers)) return;
|
|
|
|
foreach (var subscriber in subscribers)
|
|
{
|
|
try
|
|
{
|
|
// 通知订阅者
|
|
subscriber(topic, JsonUtil.FromJson<T>(message));
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
Log.Error(exception, "订阅主题 {Topic} 时发生错误", topic);
|
|
}
|
|
}
|
|
}
|
|
} |