MasstransferExporter/MasstransferInfrastructure/Mqtt/Client/MqttClient.cs

154 lines
4.2 KiB
C#
Raw Normal View History

2024-06-03 08:35:49 +00:00
using System.Security.Cryptography.X509Certificates;
using MasstransferCommon.Utils;
using MasstransferInfrastructure.Mqtt.Model;
using MasstransferSecurity.Utils;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
namespace MasstransferInfrastructure.Mqtt.Client;
class MqttClient
{
private IMqttClient? _client;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> MessageReceived;
public bool IsConnected => _client is { IsConnected: true };
/// <summary>
/// 获取连接参数
/// </summary>
/// <param name="options"></param>
/// <returns></returns>
private MqttClientOptions GetConnectionOptions(MqttConnectOptions options)
{
var clientId = DeviceInfoUtil.GenerateUniqueID();
2024-07-09 03:41:08 +00:00
// var caCert = GetCertificate(options.CaCert);
// var clientCert = GetCertificate(options.ClientCert);
//
// var chain = new X509Certificate2Collection(new[] { caCert, clientCert });
2024-06-03 08:35:49 +00:00
return new MqttClientOptionsBuilder()
.WithTcpServer(options.ServerAddress, options.Port)
2024-07-09 03:41:08 +00:00
// .WithCredentials(options.UserName, options.Password)
2024-06-03 08:35:49 +00:00
.WithClientId(clientId)
.WithCleanSession()
.WithTlsOptions(
o =>
{
o.UseTls(options.EnableTls);
o.WithSslProtocols(options.Protocols);
2024-07-09 03:41:08 +00:00
// o.WithTrustChain(chain);
2024-06-03 08:35:49 +00:00
}
)
.Build();
}
private X509Certificate2 GetCertificate(string certBase64)
{
var certBytes = Convert.FromBase64String(certBase64);
return new X509Certificate2(certBytes);
}
/// <summary>
/// 连接MQTT
/// </summary>
/// <param name="options"></param>
public async Task<bool> ConnectAsync(MqttConnectOptions options)
{
2024-07-09 03:41:08 +00:00
var ops = GetConnectionOptions(options);
2024-06-03 08:35:49 +00:00
var client = new MqttFactory().CreateMqttClient();
2024-07-09 03:41:08 +00:00
var connectResult = await client.ConnectAsync(ops);
if (connectResult.ResultCode != MqttClientConnectResultCode.Success)
2024-06-03 08:35:49 +00:00
{
return false;
}
client.ApplicationMessageReceivedAsync += (e) =>
{
MessageReceived?.Invoke(client, e);
return Task.CompletedTask;
};
_client = client;
return true;
}
/// <summary>
/// 断开连接
/// </summary>
public async Task DisconnectAsync()
{
if (_client is { IsConnected: true })
{
return;
}
await _client.DisconnectAsync();
_client = null;
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="topic"></param>
/// <param name="message"></param>
/// <param name="qos"></param>
/// <returns></returns>
public async Task<bool> Publish(string topic, object message,
MqttQualityOfServiceLevel qos)
{
if (_client is not { IsConnected: true })
{
return false;
}
var payload = message as string ?? JsonUtil.ToJson(message);
var result = await _client.PublishAsync(new MqttApplicationMessageBuilder()
2024-06-03 08:35:49 +00:00
.WithTopic(topic)
.WithPayload(payload)
.WithQualityOfServiceLevel(qos)
.Build());
return result.IsSuccess;
2024-06-03 08:35:49 +00:00
}
/// <summary>
/// 订阅主题
/// </summary>
/// <param name="topic"></param>
/// <param name="qos"></param>
/// <returns></returns>
public async Task<bool> Subscribe(string topic,
MqttQualityOfServiceLevel qos)
{
if (_client is not { IsConnected: true })
{
return false;
}
await _client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qos)
.Build());
return true;
}
/// <summary>
/// 取消订阅主题
/// </summary>
/// <param name="topic"></param>
/// <returns></returns>
public async Task<bool> Unsubscribe(string topic)
{
if (_client is not { IsConnected: true })
{
return false;
}
await _client.UnsubscribeAsync(topic);
return true;
}
}