Initial commit
This commit is contained in:
355
srcs/RobotNet10/Commons/RobotNet10.MqttConnection/MQTTClient.cs
Normal file
355
srcs/RobotNet10/Commons/RobotNet10.MqttConnection/MQTTClient.cs
Normal file
@@ -0,0 +1,355 @@
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user