356 lines
14 KiB
C#
356 lines
14 KiB
C#
using Microsoft.Extensions.Logging;
|
|
using MQTTnet;
|
|
using MQTTnet.Packets;
|
|
using MQTTnet.Protocol;
|
|
using System.Security.Cryptography.X509Certificates;
|
|
|
|
namespace RobotNet10.MqttConnection;
|
|
|
|
public class MQTTClient : IAsyncDisposable
|
|
{
|
|
private readonly MqttClientFactory MqttClientFactory;
|
|
private MqttClientOptions? MqttClientOptions;
|
|
private readonly MqttClientSubscribeOptions MqttClientSubscribeOptions;
|
|
private IMqttClient? MqttClient;
|
|
private readonly ILogger<MQTTClient> Logger;
|
|
|
|
private readonly MQTTConfig MQTTConfig;
|
|
private readonly SemaphoreSlim ReconnectionSemaphore = new(1, 1);
|
|
private volatile bool IsDisposed;
|
|
private CancellationTokenSource? cancellationConnectingTokenSource;
|
|
private CancellationTokenSource? cancellationReconnectingTokenSource;
|
|
|
|
public bool IsConnected
|
|
{
|
|
get
|
|
{
|
|
var client = MqttClient;
|
|
return !IsDisposed && client is not null && client.IsConnected;
|
|
}
|
|
}
|
|
public event Func<MqttApplicationMessageReceivedEventArgs, Task>? MessageUpdated;
|
|
|
|
public MQTTClient(MQTTConfig config, MqttTopicFilter[] topics, ILogger<MQTTClient> logger)
|
|
{
|
|
MQTTConfig = config;
|
|
Logger = logger;
|
|
|
|
MqttClientFactory = new MqttClientFactory();
|
|
var SubscribeOptionsBuilder = MqttClientFactory.CreateSubscribeOptionsBuilder();
|
|
foreach (var topic in topics)
|
|
{
|
|
SubscribeOptionsBuilder = SubscribeOptionsBuilder.WithTopicFilter(topic);
|
|
}
|
|
MqttClientSubscribeOptions = SubscribeOptionsBuilder.Build();
|
|
}
|
|
|
|
private async Task OnDisconnected(MqttClientDisconnectedEventArgs args)
|
|
{
|
|
if (IsDisposed || !args.ClientWasConnected) return;
|
|
|
|
if (!await ReconnectionSemaphore.WaitAsync(10))
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Information)) Logger.LogInformation("Reconnect is already being handled by another thread");
|
|
return;
|
|
}
|
|
|
|
try
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Warning)) Logger.LogWarning("Lost connection to the broker. Reconnection in progress...");
|
|
|
|
await CleanupCurrentClient();
|
|
|
|
await ReconnectWithRetry();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Error)) Logger.LogError("Reconnection failed: {ex}", ex.Message);
|
|
}
|
|
finally
|
|
{
|
|
ReconnectionSemaphore.Release();
|
|
}
|
|
}
|
|
|
|
private Task OnMessageReceived(MqttApplicationMessageReceivedEventArgs args)
|
|
{
|
|
try
|
|
{
|
|
if (IsDisposed) return Task.CompletedTask;
|
|
MessageUpdated?.Invoke(args);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Error)) Logger.LogError("Message Receive is failed: {Message}", ex.Message);
|
|
}
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
private async Task CleanupCurrentClient()
|
|
{
|
|
if (MqttClient is not null)
|
|
{
|
|
try
|
|
{
|
|
MqttClient.DisconnectedAsync -= OnDisconnected;
|
|
MqttClient.ApplicationMessageReceivedAsync -= OnMessageReceived;
|
|
cancellationConnectingTokenSource?.Cancel();
|
|
cancellationReconnectingTokenSource?.Cancel();
|
|
if (MqttClient.IsConnected)
|
|
{
|
|
await MqttClient.DisconnectAsync();
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Warning)) Logger.LogWarning("Cleanup client failed: {Message}", ex.Message);
|
|
}
|
|
finally
|
|
{
|
|
MqttClient.Dispose();
|
|
MqttClient = null;
|
|
cancellationConnectingTokenSource?.Dispose();
|
|
cancellationReconnectingTokenSource?.Dispose();
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task ReconnectWithRetry()
|
|
{
|
|
const int maxRetries = 5;
|
|
const int retryDelayMs = 3000;
|
|
|
|
for (int attempt = 1; attempt <= maxRetries && !IsDisposed; attempt++)
|
|
{
|
|
cancellationReconnectingTokenSource?.Dispose();
|
|
cancellationReconnectingTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
|
try
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Information)) Logger.LogInformation("Reconnection attempt {attempt}/{maxRetries}", attempt, maxRetries);
|
|
|
|
await ConnectAsync(cancellationReconnectingTokenSource.Token);
|
|
|
|
if (IsConnected)
|
|
{
|
|
await SubscribeAsync(cancellationReconnectingTokenSource.Token);
|
|
if (Logger.IsEnabled(LogLevel.Information)) Logger.LogInformation("Reconnection successfully");
|
|
return;
|
|
}
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Warning)) Logger.LogWarning("Reconnection attempt {tempt} timed out", attempt);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Error)) Logger.LogError("Reconnect attempt {tempt} failed: {Message}", attempt, ex.Message);
|
|
}
|
|
|
|
if (attempt < maxRetries && !IsDisposed)
|
|
{
|
|
try
|
|
{
|
|
await Task.Delay(retryDelayMs * attempt, cancellationReconnectingTokenSource.Token);
|
|
}
|
|
catch { }
|
|
}
|
|
}
|
|
|
|
Logger.LogError("Không thể reconnect sau tất cả các attempts");
|
|
}
|
|
|
|
private bool ValidateCertificates(MqttClientCertificateValidationEventArgs arg)
|
|
{
|
|
if (!string.IsNullOrEmpty(MQTTConfig.CaCertificatesPath))
|
|
{
|
|
if (File.Exists(MQTTConfig.CaCertificatesPath))
|
|
{
|
|
var caCert = X509CertificateLoader.LoadCertificateFromFile(MQTTConfig.CaCertificatesPath);
|
|
arg.Chain.ChainPolicy.ExtraStore.Add(caCert);
|
|
arg.Chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;
|
|
arg.Chain.ChainPolicy.VerificationFlags = X509VerificationFlags.AllowUnknownCertificateAuthority;
|
|
|
|
return arg.Chain.Build((X509Certificate2)arg.Certificate);
|
|
}
|
|
}
|
|
return !MQTTConfig.EnableCA;
|
|
}
|
|
|
|
private void BuildMqttClientOptions()
|
|
{
|
|
var builder = MqttClientFactory.CreateClientOptionsBuilder()
|
|
.WithTcpServer(MQTTConfig.Host, MQTTConfig.Port)
|
|
.WithClientId($"{MQTTConfig.ClientId}_{Guid.NewGuid()}")
|
|
.WithCleanSession(true);
|
|
if (MQTTConfig.EnablePassword)
|
|
{
|
|
builder = builder.WithCredentials(MQTTConfig.Username, MQTTConfig.Password);
|
|
}
|
|
|
|
if (MQTTConfig.EnableTls)
|
|
{
|
|
var tlsOptionsBuilder = new MqttClientTlsOptionsBuilder()
|
|
.UseTls(true)
|
|
.WithCertificateValidationHandler(ValidateCertificates)
|
|
.WithClientCertificatesProvider(new MQTTClientCertificatesProvider(MQTTConfig.ClientCertificatePath, MQTTConfig.ClientKeyPath));
|
|
builder = builder.WithTlsOptions(tlsOptionsBuilder.Build());
|
|
}
|
|
MqttClientOptions = builder.Build();
|
|
}
|
|
|
|
public async Task ConnectAsync(CancellationToken? cancellationToken)
|
|
{
|
|
if (!IsDisposed)
|
|
{
|
|
BuildMqttClientOptions();
|
|
await CleanupCurrentClient();
|
|
|
|
MqttClient = MqttClientFactory.CreateMqttClient();
|
|
|
|
MqttClient.ApplicationMessageReceivedAsync -= OnMessageReceived;
|
|
MqttClient.ApplicationMessageReceivedAsync += OnMessageReceived;
|
|
MqttClient.DisconnectedAsync -= OnDisconnected;
|
|
MqttClient.DisconnectedAsync += OnDisconnected;
|
|
|
|
cancellationConnectingTokenSource?.Dispose();
|
|
cancellationConnectingTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken ?? CancellationToken.None);
|
|
while (!cancellationConnectingTokenSource.IsCancellationRequested && !IsDisposed)
|
|
{
|
|
try
|
|
{
|
|
var connection = await MqttClient.ConnectAsync(MqttClientOptions, cancellationConnectingTokenSource.Token);
|
|
if (connection.ResultCode != MqttClientConnectResultCode.Success || !MqttClient.IsConnected)
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Warning)) Logger.LogWarning("Connection to broker failed: {ReasonString}", connection.ReasonString);
|
|
}
|
|
else
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Information)) Logger.LogInformation("Connected to {Host} successfully", MQTTConfig.Host);
|
|
break;
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Error)) Logger.LogError("Create MQTT Client failed: {ex}", ex.Message);
|
|
}
|
|
try
|
|
{
|
|
await Task.Delay(3000, cancellationConnectingTokenSource.Token);
|
|
}
|
|
catch { }
|
|
}
|
|
}
|
|
else throw new ObjectDisposedException(nameof(MQTTClient));
|
|
}
|
|
|
|
public async Task SubscribeAsync(CancellationToken? cancellationToken)
|
|
{
|
|
if (!IsDisposed)
|
|
{
|
|
if (MqttClient is null) throw new Exception("Attempted to subscribe before broker connection was initialized");
|
|
if (!MqttClient.IsConnected) throw new Exception("Attempted to subscribe while connection to broker is not successful");
|
|
|
|
cancellationConnectingTokenSource?.Dispose();
|
|
cancellationConnectingTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken ?? CancellationToken.None);
|
|
while (!cancellationConnectingTokenSource.IsCancellationRequested && !IsDisposed)
|
|
{
|
|
try
|
|
{
|
|
var response = await MqttClient.SubscribeAsync(MqttClientSubscribeOptions, cancellationConnectingTokenSource.Token);
|
|
bool isSuccess = true;
|
|
foreach (var item in response.Items)
|
|
{
|
|
if (item.ResultCode == MqttClientSubscribeResultCode.GrantedQoS0 ||
|
|
item.ResultCode == MqttClientSubscribeResultCode.GrantedQoS1 ||
|
|
item.ResultCode == MqttClientSubscribeResultCode.GrantedQoS2)
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Information)) Logger.LogInformation("Subscribed to topic '{Topic}' with granted QoS {ResultCode}", item.TopicFilter.Topic, item.ResultCode);
|
|
}
|
|
else
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Warning)) Logger.LogWarning("Subscribe to {Topic} failed with reason: {res}", item.TopicFilter.Topic, response.ReasonString);
|
|
isSuccess = false;
|
|
break;
|
|
}
|
|
}
|
|
if (isSuccess) break;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Error)) Logger.LogError("Subscribe failed: {ex}", ex.Message);
|
|
}
|
|
if (!cancellationConnectingTokenSource.IsCancellationRequested && !IsDisposed)
|
|
{
|
|
try
|
|
{
|
|
await Task.Delay(3000, cancellationConnectingTokenSource.Token);
|
|
}
|
|
catch { }
|
|
}
|
|
}
|
|
}
|
|
else throw new ObjectDisposedException(nameof(MQTTClient));
|
|
}
|
|
|
|
public async Task PublishAsync(
|
|
string topic,
|
|
string data,
|
|
MqttQualityOfServiceLevel QoS = MqttQualityOfServiceLevel.AtLeastOnce,
|
|
bool retain = false)
|
|
{
|
|
if (IsDisposed) throw new Exception("Client has been disposed");
|
|
var repeat = MQTTConfig.PublishRepeat;
|
|
while (repeat-- > 0 && !IsDisposed)
|
|
{
|
|
try
|
|
{
|
|
var applicationMessage = MqttClientFactory.CreateApplicationMessageBuilder()
|
|
.WithTopic(topic)
|
|
.WithPayload(data)
|
|
.WithQualityOfServiceLevel(QoS)
|
|
.WithRetainFlag(retain)
|
|
.Build();
|
|
if (MqttClient is null || !IsConnected) throw new Exception("Not connected to the broker");
|
|
var publish = await MqttClient.PublishAsync(applicationMessage);
|
|
if (!publish.IsSuccess)
|
|
{
|
|
await Task.Delay(500);
|
|
continue;
|
|
}
|
|
return;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
if (Logger.IsEnabled(LogLevel.Error)) Logger.LogError("Publish failed: {ex}", ex.Message);
|
|
}
|
|
}
|
|
throw new Exception("Cannot publish message to broker");
|
|
}
|
|
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
if (IsDisposed) return;
|
|
IsDisposed = true;
|
|
cancellationConnectingTokenSource?.Cancel();
|
|
cancellationReconnectingTokenSource?.Cancel();
|
|
try
|
|
{
|
|
if (!await ReconnectionSemaphore.WaitAsync(TimeSpan.FromSeconds(2)))
|
|
{
|
|
Logger.LogWarning("Failed to acquire semaphore during dispose, forcing cleanup");
|
|
return;
|
|
}
|
|
|
|
await CleanupCurrentClient();
|
|
}
|
|
finally
|
|
{
|
|
ReconnectionSemaphore.Release();
|
|
ReconnectionSemaphore.Dispose();
|
|
cancellationConnectingTokenSource?.Dispose();
|
|
cancellationReconnectingTokenSource?.Dispose();
|
|
GC.SuppressFinalize(this);
|
|
}
|
|
}
|
|
}
|