29408474270

Committed 15 Jul 2026 10:32AM UTC coverage: 85.051% (+1.5%) from 83.551%

Build # 29408474270

Build Type

Pull #1385

github

Committed by

web-flow

Commit Message

Merge 142d16019 into bf6c3500c

Pull Request Pull Request #1385: MQTT Upgrade - attempt 2

Coverage Stats

977 of 1293 branches covered (75.56%)

Branch coverage included in aggregate %.

220 of 248 new or added lines in 5 files covered. (88.71%)

4 existing lines in 3 files now uncovered.

3660 of 4159 relevant lines covered (88.0%)

137.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.5

/src/Extensions/NetDaemon.Extensions.MqttEntityManager/AssuredMqttConnection.cs

using System.Collections.Concurrent;
using System.Globalization;
using System.Threading.Channels;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using MQTTnet;
using MQTTnet.Packets;
using NetDaemon.Extensions.MqttEntityManager.Exceptions;

namespace NetDaemon.Extensions.MqttEntityManager;

/// <summary>
/// Wrapper to assure an MQTT connection.
/// </summary>
internal class AssuredMqttConnection : IAssuredMqttConnection, IDisposable
{
    private static readonly TimeSpan DefaultReconnectDelay = TimeSpan.FromSeconds(5);
    private static readonly TimeSpan DefaultConnectedPollDelay = TimeSpan.FromSeconds(1);
    private static readonly TimeSpan PublishRetryDelay = TimeSpan.FromSeconds(1);
    private const int MaxConnectedPublishAttempts = 3;

    private readonly ILogger<AssuredMqttConnection> _logger;
    private readonly IMqttClient _mqttClient;
    private readonly MqttClientOptions _clientOptions;
    private readonly TimeProvider _timeProvider;
    private readonly Channel<MqttApplicationMessage> _publishQueue = Channel.CreateUnbounded<MqttApplicationMessage>(
        new UnboundedChannelOptions
        {
            SingleReader = true,
            SingleWriter = false
        });
    private readonly ConcurrentDictionary<string, MqttTopicFilter> _subscriptions = new();
    private readonly ConcurrentDictionary<string, MqttTopicFilter> _pendingSubscriptions = new();
    private readonly CancellationTokenSource _stopping = new();
    private readonly SemaphoreSlim _clientOperationGate = new(1, 1);
    private readonly object _connectionStateLock = new();
    private readonly Task _connectionTask;
    private readonly Task _publishTask;
    private TaskCompletionSource _connectedSignal = CreateConnectionSignal();
    private bool _disposed;
    private volatile bool _hasConnected;

    /// <summary>
    /// Initializes a new instance of the <see cref="AssuredMqttConnection"/> class.
    /// </summary>
    /// <param name="logger">The logger.</param>
    /// <param name="mqttClientOptionsFactory">The MQTT client options factory.</param>
    /// <param name="mqttFactory">The MQTT factory wrapper.</param>
    /// <param name="mqttConfig">The MQTT configuration.</param>
    /// <param name="timeProvider">The clock used for connection retry delays.</param>
    public AssuredMqttConnection(
        ILogger<AssuredMqttConnection> logger,
        IMqttClientOptionsFactory mqttClientOptionsFactory,
        IMqttFactory mqttFactory,
        IOptions<MqttConfiguration> mqttConfig,
        TimeProvider? timeProvider = null)
    {
        _logger = logger;
        _timeProvider = timeProvider ?? TimeProvider.System;

        _logger.LogTrace("MQTT initiating connection");
        _clientOptions = mqttClientOptionsFactory.CreateClientOptions(mqttConfig.Value);
        _mqttClient = mqttFactory.CreateMqttClient();

        _mqttClient.ConnectedAsync += MqttClientOnConnectedAsync;
        _mqttClient.DisconnectedAsync += MqttClientOnDisconnectedAsync;
        _mqttClient.ApplicationMessageReceivedAsync += MqttClientOnApplicationMessageReceivedAsync;

        _connectionTask = Task.Run(() => MaintainConnectionAsync(_stopping.Token));
        _publishTask = Task.Run(() => PublishQueuedMessagesAsync(_stopping.Token));
    }

    /// <inheritdoc />
    public event Func<MqttApplicationMessageReceivedEventArgs, Task>? ApplicationMessageReceivedAsync;

    /// <inheritdoc />
    public async Task PublishAsync(MqttApplicationMessage message)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        await _publishQueue.Writer.WriteAsync(message, _stopping.Token).ConfigureAwait(false);
    }

    /// <inheritdoc />
    public async Task SubscribeAsync(MqttTopicFilter topicFilter)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);

        if (string.IsNullOrEmpty(topicFilter.Topic))
        {
            throw new ArgumentException("MQTT topic filter must specify a topic.", nameof(topicFilter));
        }

        _subscriptions[topicFilter.Topic] = topicFilter;
        _pendingSubscriptions[topicFilter.Topic] = topicFilter;

        if (_mqttClient.IsConnected)
        {
            await TrySubscribePendingAsync(topicFilter.Topic, topicFilter, _stopping.Token).ConfigureAwait(false);
        }
    }

    private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            try
            {
                if (_mqttClient.IsConnected)
                {
                    await SubscribePendingAsync(cancellationToken).ConfigureAwait(false);
                    await Task.Delay(DefaultConnectedPollDelay, _timeProvider, cancellationToken).ConfigureAwait(false);
                    continue;
                }

                MarkConnectionLost();

                _logger.LogTrace("Connecting to MQTT broker at {Host}:{Port}/{UserName}",
                    GetConfiguredHost(), GetConfiguredPort(), _clientOptions.Credentials?.GetUserName(_clientOptions));

                var connectResult = await ExecuteClientOperationAsync(
                    operationToken => _mqttClient.ConnectAsync(_clientOptions, operationToken),
                    cancellationToken).ConfigureAwait(false);

                if (connectResult.ResultCode != MqttClientConnectResultCode.Success)
                {
                    _logger.LogTrace("MQTT connection rejected: {ResultCode} {ReasonString}",
                        connectResult.ResultCode, connectResult.ReasonString);
                    await Task.Delay(DefaultReconnectDelay, _timeProvider, cancellationToken).ConfigureAwait(false);
                    continue;
                }

                _logger.LogTrace("MQTT client is ready");
                _hasConnected = true;
                MarkConnected();
                MarkAllSubscriptionsPending();
                await SubscribePendingAsync(cancellationToken).ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                _logger.LogTrace(ex, "MQTT connection attempt failed");
                await DelayReconnectAsync(cancellationToken).ConfigureAwait(false);
            }
        }
    }

    private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
    {
        await foreach (var message in _publishQueue.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
        {
            var connectedFailureCount = 0;
            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    if (!_mqttClient.IsConnected)
                    {
                        await WaitForConnectionAsync(cancellationToken).ConfigureAwait(false);
                        continue;
                    }

                    var published = await ExecuteClientOperationAsync(async operationToken =>
                    {
                        if (!_mqttClient.IsConnected)
                        {
                            return false;
                        }

                        var publishResult = await _mqttClient.PublishAsync(message, operationToken).ConfigureAwait(false);
                        if (!publishResult.IsSuccess)
                        {
                            throw new MqttPublishException(
                                $"MQTT broker rejected the message with {publishResult.ReasonCode}: {publishResult.ReasonString}");
                        }

                        return true;
                    }, cancellationToken).ConfigureAwait(false);

                    if (!published)
                    {
                        MarkConnectionLost();
                        continue;
                    }

                    break;
                }
                catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
                {
                    break;
                }
                catch (Exception ex)
                {
                    if (!_mqttClient.IsConnected)
                    {
                        connectedFailureCount = 0;
                        MarkConnectionLost();
                        _logger.LogDebug(ex, "Failed to publish MQTT message. The message will be retried after reconnect.");
                        await WaitForConnectionAsync(cancellationToken).ConfigureAwait(false);
                        continue;
                    }

                    connectedFailureCount++;
                    if (connectedFailureCount >= MaxConnectedPublishAttempts)
                    {
                        _logger.LogError(ex,
                            "Discarding MQTT message for topic {Topic} after {AttemptCount} failed publish attempts while connected.",
                            message.Topic, connectedFailureCount);
                        break;
                    }

                    _logger.LogDebug(ex,
                        "Failed to publish MQTT message while connected. Retrying attempt {NextAttempt} of {MaxAttempts} after a delay.",
                        connectedFailureCount + 1, MaxConnectedPublishAttempts);
                    await Task.Delay(PublishRetryDelay, _timeProvider, cancellationToken).ConfigureAwait(false);
                }
            }
        }
    }

    private async Task SubscribePendingAsync(CancellationToken cancellationToken)
    {
        foreach (var (topic, topicFilter) in _pendingSubscriptions)
        {
            if (!_mqttClient.IsConnected)
            {
                return;
            }

            await TrySubscribePendingAsync(topic, topicFilter, cancellationToken).ConfigureAwait(false);
        }
    }

    private async Task TrySubscribePendingAsync(
        string topic,
        MqttTopicFilter topicFilter,
        CancellationToken cancellationToken)
    {
        try
        {
            var subscribed = await ExecuteClientOperationAsync(async operationToken =>
            {
                if (!_mqttClient.IsConnected ||
                    !_pendingSubscriptions.TryGetValue(topic, out var currentTopicFilter) ||
                    !ReferenceEquals(currentTopicFilter, topicFilter))
                {
                    return false;
                }

                var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
                    .WithTopicFilter(topicFilter)
                    .Build();

                var subscribeResult = await _mqttClient.SubscribeAsync(subscribeOptions, operationToken).ConfigureAwait(false);
                var accepted = subscribeResult.Items.Count > 0 && subscribeResult.Items.All(IsSuccessfulSubscribeResult);
                if (!accepted)
                {
                    _logger.LogDebug(
                        "MQTT broker rejected subscription to topic {Topic}: {ResultCodes} {ReasonString}",
                        topic,
                        string.Join(", ", subscribeResult.Items.Select(item => item.ResultCode)),
                        subscribeResult.ReasonString);
                }

                return accepted && _mqttClient.IsConnected;
            }, cancellationToken).ConfigureAwait(false);

            if (subscribed &&
                _pendingSubscriptions.TryGetValue(topic, out var currentTopicFilter) &&
                ReferenceEquals(currentTopicFilter, topicFilter))
            {
                _pendingSubscriptions.TryRemove(topic, out _);
            }
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw;
        }
        catch (Exception ex)
        {
            _logger.LogDebug(ex, "Failed to subscribe to MQTT topic {Topic}. The subscription will be retried.", topic);
        }
    }

    private async Task<T> ExecuteClientOperationAsync<T>(
        Func<CancellationToken, Task<T>> operation,
        CancellationToken cancellationToken)
    {
        await _clientOperationGate.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            using var operationTimeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            operationTimeout.CancelAfter(_clientOptions.Timeout);
            return await operation(operationTimeout.Token).ConfigureAwait(false);
        }
        finally
        {
            _clientOperationGate.Release();
        }
    }

    private void MarkAllSubscriptionsPending()
    {
        foreach (var (topic, topicFilter) in _subscriptions)
        {
            _pendingSubscriptions[topic] = topicFilter;
        }
    }

    private static bool IsSuccessfulSubscribeResult(MqttClientSubscribeResultItem item)
    {
        return item.ResultCode is MqttClientSubscribeResultCode.GrantedQoS0
            or MqttClientSubscribeResultCode.GrantedQoS1
            or MqttClientSubscribeResultCode.GrantedQoS2;
    }

    private Task MqttClientOnDisconnectedAsync(MqttClientDisconnectedEventArgs arg)
    {
        if (_disposed || _stopping.IsCancellationRequested)
        {
            return Task.CompletedTask;
        }

        if (_hasConnected)
        {
            _logger.LogDebug("MQTT disconnected: {Reason}", BuildErrorResponse(arg));
        }
        else
        {
            _logger.LogTrace("MQTT disconnected before the initial connection completed: {Reason}", BuildErrorResponse(arg));
        }

        MarkConnectionLost();
        return Task.CompletedTask;
    }

    private Task MqttClientOnConnectedAsync(MqttClientConnectedEventArgs arg)
    {
        _hasConnected = true;
        _logger.LogDebug("MQTT connected: {ResultCode}", arg.ConnectResult.ResultCode);
        MarkConnected();
        return Task.CompletedTask;
    }

    private async Task MqttClientOnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
    {
        var handlers = ApplicationMessageReceivedAsync;
        if (handlers is null)
        {
            return;
        }

        foreach (Func<MqttApplicationMessageReceivedEventArgs, Task> handler in handlers.GetInvocationList())
        {
            await handler(arg).ConfigureAwait(false);
        }
    }

    private static string BuildErrorResponse(MqttClientDisconnectedEventArgs arg)
    {
        var sb = new System.Text.StringBuilder();

        sb.AppendLine(CultureInfo.InvariantCulture, $"{arg.Exception?.Message} ({arg.Reason})");
        var ex = arg.Exception?.InnerException;
        while (ex != null)
        {
            sb.AppendLine(ex.Message);
            ex = ex.InnerException;
        }

        return sb.ToString();
    }

    private static TaskCompletionSource CreateConnectionSignal()
    {
        return new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
    }

    private void MarkConnected()
    {
        lock (_connectionStateLock)
        {
            _connectedSignal.TrySetResult();
        }
    }

    private void MarkConnectionLost()
    {
        lock (_connectionStateLock)
        {
            if (_connectedSignal.Task.IsCompleted)
            {
                _connectedSignal = CreateConnectionSignal();
            }
        }

        MarkAllSubscriptionsPending();
    }

    private async Task WaitForConnectionAsync(CancellationToken cancellationToken)
    {
        while (!_mqttClient.IsConnected)
        {
            Task connectedTask;
            lock (_connectionStateLock)
            {
                if (_mqttClient.IsConnected)
                {
                    return;
                }

                connectedTask = _connectedSignal.Task;
            }

            await connectedTask.WaitAsync(cancellationToken).ConfigureAwait(false);
        }
    }

    private async Task DelayReconnectAsync(CancellationToken cancellationToken)
    {
        try
        {
            await Task.Delay(DefaultReconnectDelay, _timeProvider, cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
        }
    }

    private string? GetConfiguredHost()
    {
        return (_clientOptions.ChannelOptions as MqttClientTcpOptions)?.RemoteEndpoint switch
        {
            System.Net.DnsEndPoint dnsEndPoint => dnsEndPoint.Host,
            System.Net.IPEndPoint ipEndPoint => ipEndPoint.Address.ToString(),
            _ => null
        };
    }

    private int? GetConfiguredPort()
    {
        return (_clientOptions.ChannelOptions as MqttClientTcpOptions)?.RemoteEndpoint switch
        {
            System.Net.DnsEndPoint dnsEndPoint => dnsEndPoint.Port,
            System.Net.IPEndPoint ipEndPoint => ipEndPoint.Port,
            _ => null
        };
    }

    /// <inheritdoc />
    public void Dispose()
    {
        if (_disposed)
        {
            return;
        }

        _disposed = true;
        _logger.LogTrace("MQTT disconnecting");
        _stopping.Cancel();
        _publishQueue.Writer.TryComplete();

        try
        {
            Task.WaitAll([_connectionTask, _publishTask], TimeSpan.FromSeconds(1));
        }
        catch (AggregateException)
        {
        }

        _stopping.Dispose();
        _clientOperationGate.Dispose();

        if (_mqttClient is IDisposable disposable)
        {
            disposable.Dispose();
        }
    }
}

1	using System.Collections.Concurrent;
2	using System.Globalization;
3	using System.Threading.Channels;
4	using Microsoft.Extensions.Logging;
5	using Microsoft.Extensions.Options;
6	using MQTTnet;
7	using MQTTnet.Packets;
8	using NetDaemon.Extensions.MqttEntityManager.Exceptions;
9
10	namespace NetDaemon.Extensions.MqttEntityManager;
11
12	/// <summary>
13	/// Wrapper to assure an MQTT connection.
14	/// </summary>
15	internal class AssuredMqttConnection : IAssuredMqttConnection, IDisposable
16	{
17	private static readonly TimeSpan DefaultReconnectDelay = TimeSpan.FromSeconds(5);	3✔
18	private static readonly TimeSpan DefaultConnectedPollDelay = TimeSpan.FromSeconds(1);	3✔
19	private static readonly TimeSpan PublishRetryDelay = TimeSpan.FromSeconds(1);	3✔
20	private const int MaxConnectedPublishAttempts = 3;
21
22	private readonly ILogger<AssuredMqttConnection> _logger;
23	private readonly IMqttClient _mqttClient;
24	private readonly MqttClientOptions _clientOptions;
25	private readonly TimeProvider _timeProvider;
26	private readonly Channel<MqttApplicationMessage> _publishQueue = Channel.CreateUnbounded<MqttApplicationMessage>(	21✔
27	new UnboundedChannelOptions	21✔
28	{	21✔
29	SingleReader = true,	21✔
30	SingleWriter = false	21✔
31	});	21✔
32	private readonly ConcurrentDictionary<string, MqttTopicFilter> _subscriptions = new();	21✔
33	private readonly ConcurrentDictionary<string, MqttTopicFilter> _pendingSubscriptions = new();	21✔
34	private readonly CancellationTokenSource _stopping = new();	21✔
35	private readonly SemaphoreSlim _clientOperationGate = new(1, 1);	21✔
36	private readonly object _connectionStateLock = new();	21✔
37	private readonly Task _connectionTask;
38	private readonly Task _publishTask;
39	private TaskCompletionSource _connectedSignal = CreateConnectionSignal();	21✔
40	private bool _disposed;
41	private volatile bool _hasConnected;
42
43	/// <summary>
44	/// Initializes a new instance of the <see cref="AssuredMqttConnection"/> class.
45	/// </summary>
46	/// <param name="logger">The logger.</param>
47	/// <param name="mqttClientOptionsFactory">The MQTT client options factory.</param>
48	/// <param name="mqttFactory">The MQTT factory wrapper.</param>
49	/// <param name="mqttConfig">The MQTT configuration.</param>
50	/// <param name="timeProvider">The clock used for connection retry delays.</param>
51	public AssuredMqttConnection(	21✔
52	ILogger<AssuredMqttConnection> logger,	21✔
53	IMqttClientOptionsFactory mqttClientOptionsFactory,	21✔
54	IMqttFactory mqttFactory,	21✔
55	IOptions<MqttConfiguration> mqttConfig,	21✔
56	TimeProvider? timeProvider = null)	21✔
57	{
58	_logger = logger;	21✔
59	_timeProvider = timeProvider ?? TimeProvider.System;	21✔
60
61	_logger.LogTrace("MQTT initiating connection");	21✔
62	_clientOptions = mqttClientOptionsFactory.CreateClientOptions(mqttConfig.Value);	21✔
63	_mqttClient = mqttFactory.CreateMqttClient();	21✔
64
65	_mqttClient.ConnectedAsync += MqttClientOnConnectedAsync;	21✔
66	_mqttClient.DisconnectedAsync += MqttClientOnDisconnectedAsync;	21✔
67	_mqttClient.ApplicationMessageReceivedAsync += MqttClientOnApplicationMessageReceivedAsync;	21✔
68
69	_connectionTask = Task.Run(() => MaintainConnectionAsync(_stopping.Token));	42✔
70	_publishTask = Task.Run(() => PublishQueuedMessagesAsync(_stopping.Token));	42✔
71	}	21✔
72
73	/// <inheritdoc />
74	public event Func<MqttApplicationMessageReceivedEventArgs, Task>? ApplicationMessageReceivedAsync;
75
76	/// <inheritdoc />
77	public async Task PublishAsync(MqttApplicationMessage message)
78	{
79	ObjectDisposedException.ThrowIf(_disposed, this);	31✔
80	await _publishQueue.Writer.WriteAsync(message, _stopping.Token).ConfigureAwait(false);	31✔
81	}	31✔
82
83	/// <inheritdoc />
84	public async Task SubscribeAsync(MqttTopicFilter topicFilter)
85	{
86	ObjectDisposedException.ThrowIf(_disposed, this);	16✔
87
88	if (string.IsNullOrEmpty(topicFilter.Topic))	16!
89	{
NEW 90	throw new ArgumentException("MQTT topic filter must specify a topic.", nameof(topicFilter));	×
91	}
92
93	_subscriptions[topicFilter.Topic] = topicFilter;	16✔
94	_pendingSubscriptions[topicFilter.Topic] = topicFilter;	16✔
95
96	if (_mqttClient.IsConnected)	16✔
97	{
98	await TrySubscribePendingAsync(topicFilter.Topic, topicFilter, _stopping.Token).ConfigureAwait(false);	6✔
99	}
100	}	16✔
101
102	private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
103	{
104	while (!cancellationToken.IsCancellationRequested)	62✔
105	{
106	try
107	{
108	if (_mqttClient.IsConnected)	61✔
109	{
110	await SubscribePendingAsync(cancellationToken).ConfigureAwait(false);	34✔
111	await Task.Delay(DefaultConnectedPollDelay, _timeProvider, cancellationToken).ConfigureAwait(false);	34✔
112	continue;	17✔
113	}
114
115	MarkConnectionLost();	27✔
116
117	_logger.LogTrace("Connecting to MQTT broker at {Host}:{Port}/{UserName}",	27!
118	GetConfiguredHost(), GetConfiguredPort(), _clientOptions.Credentials?.GetUserName(_clientOptions));	27✔
119
120	var connectResult = await ExecuteClientOperationAsync(	27✔
121	operationToken => _mqttClient.ConnectAsync(_clientOptions, operationToken),	27✔
122	cancellationToken).ConfigureAwait(false);	27✔
123
124	if (connectResult.ResultCode != MqttClientConnectResultCode.Success)	22!
125	{
NEW 126	_logger.LogTrace("MQTT connection rejected: {ResultCode} {ReasonString}",	×
NEW 127	connectResult.ResultCode, connectResult.ReasonString);	×
NEW 128	await Task.Delay(DefaultReconnectDelay, _timeProvider, cancellationToken).ConfigureAwait(false);	×
NEW 129	continue;	×
130	}
131
132	_logger.LogTrace("MQTT client is ready");	22✔
133	_hasConnected = true;	22✔
134	MarkConnected();	22✔
135	MarkAllSubscriptionsPending();	22✔
136	await SubscribePendingAsync(cancellationToken).ConfigureAwait(false);	22✔
137	}	22✔
138	catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)	11✔
139	{
140	break;	10✔
141	}
142	catch (Exception ex)	5✔
143	{
144	_logger.LogTrace(ex, "MQTT connection attempt failed");	5✔
145	await DelayReconnectAsync(cancellationToken).ConfigureAwait(false);	5✔
146	}
147	}
148	}	11✔
149
150	private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
151	{
152	await foreach (var message in _publishQueue.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))	104!
153	{
154	var connectedFailureCount = 0;	31✔
155	while (!cancellationToken.IsCancellationRequested)	43✔
156	{
157	try
158	{
159	if (!_mqttClient.IsConnected)	43✔
160	{
161	await WaitForConnectionAsync(cancellationToken).ConfigureAwait(false);	8✔
162	continue;	8✔
163	}
164
165	var published = await ExecuteClientOperationAsync(async operationToken =>	35✔
166	{	35✔
167	if (!_mqttClient.IsConnected)	35!
168	{	35✔
NEW 169	return false;	×
170	}	35✔
171		35✔
172	var publishResult = await _mqttClient.PublishAsync(message, operationToken).ConfigureAwait(false);	35✔
173	if (!publishResult.IsSuccess)	32✔
174	{	35✔
175	throw new MqttPublishException(	3✔
176	$"MQTT broker rejected the message with {publishResult.ReasonCode}: {publishResult.ReasonString}");	3✔
177	}	35✔
178		35✔
179	return true;	29✔
180	}, cancellationToken).ConfigureAwait(false);	64✔
181
182	if (!published)	29!
183	{
NEW 184	MarkConnectionLost();	×
NEW 185	continue;	×
186	}
187
188	break;	29✔
189	}
NEW 190	catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)	×
191	{
NEW 192	break;	×
193	}
194	catch (Exception ex)
195	{
196	if (!_mqttClient.IsConnected)	6!
197	{
NEW 198	connectedFailureCount = 0;	×
NEW 199	MarkConnectionLost();	×
NEW 200	_logger.LogDebug(ex, "Failed to publish MQTT message. The message will be retried after reconnect.");	×
NEW 201	await WaitForConnectionAsync(cancellationToken).ConfigureAwait(false);	×
NEW 202	continue;	×
203	}
204
205	connectedFailureCount++;	6✔
206	if (connectedFailureCount >= MaxConnectedPublishAttempts)	6✔
207	{
208	_logger.LogError(ex,	2✔
209	"Discarding MQTT message for topic {Topic} after {AttemptCount} failed publish attempts while connected.",	2✔
210	message.Topic, connectedFailureCount);	2✔
211	break;	2✔
212	}
213
214	_logger.LogDebug(ex,	4✔
215	"Failed to publish MQTT message while connected. Retrying attempt {NextAttempt} of {MaxAttempts} after a delay.",	4✔
216	connectedFailureCount + 1, MaxConnectedPublishAttempts);	4✔
217	await Task.Delay(PublishRetryDelay, _timeProvider, cancellationToken).ConfigureAwait(false);	4✔
218	}
219	}
220	}	31✔
NEW 221	}	×
222
223	private async Task SubscribePendingAsync(CancellationToken cancellationToken)
224	{
225	foreach (var (topic, topicFilter) in _pendingSubscriptions)	142✔
226	{
227	if (!_mqttClient.IsConnected)	15!
228	{
NEW 229	return;	×
230	}
231
232	await TrySubscribePendingAsync(topic, topicFilter, cancellationToken).ConfigureAwait(false);	15✔
233	}
234	}	56✔
235
236	private async Task TrySubscribePendingAsync(
237	string topic,
238	MqttTopicFilter topicFilter,
239	CancellationToken cancellationToken)
240	{
241	try
242	{
243	var subscribed = await ExecuteClientOperationAsync(async operationToken =>	21✔
244	{	21✔
245	if (!_mqttClient.IsConnected \|\|	21✔
246	!_pendingSubscriptions.TryGetValue(topic, out var currentTopicFilter) \|\|	21✔
247	!ReferenceEquals(currentTopicFilter, topicFilter))	21✔
248	{	21✔
249	return false;	1✔
250	}	21✔
251		21✔
252	var subscribeOptions = new MqttClientSubscribeOptionsBuilder()	20✔
253	.WithTopicFilter(topicFilter)	20✔
254	.Build();	20✔
255		21✔
256	var subscribeResult = await _mqttClient.SubscribeAsync(subscribeOptions, operationToken).ConfigureAwait(false);	20✔
257	var accepted = subscribeResult.Items.Count > 0 && subscribeResult.Items.All(IsSuccessfulSubscribeResult);	19!
258	if (!accepted)	19✔
259	{	21✔
260	_logger.LogDebug(	1✔
261	"MQTT broker rejected subscription to topic {Topic}: {ResultCodes} {ReasonString}",	1✔
262	topic,	1✔
263	string.Join(", ", subscribeResult.Items.Select(item => item.ResultCode)),	1✔
264	subscribeResult.ReasonString);	1✔
265	}	21✔
266		21✔
267	return accepted && _mqttClient.IsConnected;	19✔
268	}, cancellationToken).ConfigureAwait(false);	41✔
269
270	if (subscribed &&	20✔
271	_pendingSubscriptions.TryGetValue(topic, out var currentTopicFilter) &&	20✔
272	ReferenceEquals(currentTopicFilter, topicFilter))	20✔
273	{
274	_pendingSubscriptions.TryRemove(topic, out _);	17✔
275	}
276	}	20✔
NEW 277	catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)	×
278	{
NEW 279	throw;	×
280	}
281	catch (Exception ex)	1✔
282	{
283	_logger.LogDebug(ex, "Failed to subscribe to MQTT topic {Topic}. The subscription will be retried.", topic);	1✔
284	}	1✔
285	}	21✔
286
287	private async Task<T> ExecuteClientOperationAsync<T>(
288	Func<CancellationToken, Task<T>> operation,
289	CancellationToken cancellationToken)
290	{
291	await _clientOperationGate.WaitAsync(cancellationToken).ConfigureAwait(false);	83✔
292	try
293	{
294	using var operationTimeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);	83✔
295	operationTimeout.CancelAfter(_clientOptions.Timeout);	83✔
296	return await operation(operationTimeout.Token).ConfigureAwait(false);	83✔
297	}
298	finally
299	{
300	_clientOperationGate.Release();	83✔
301	}
302	}	71✔
303
304	private void MarkAllSubscriptionsPending()
305	{
306	foreach (var (topic, topicFilter) in _subscriptions)	196✔
307	{
308	_pendingSubscriptions[topic] = topicFilter;	36✔
309	}
310	}	62✔
311
312	private static bool IsSuccessfulSubscribeResult(MqttClientSubscribeResultItem item)
313	{
314	return item.ResultCode is MqttClientSubscribeResultCode.GrantedQoS0	19✔
315	or MqttClientSubscribeResultCode.GrantedQoS1	19✔
316	or MqttClientSubscribeResultCode.GrantedQoS2;	19✔
317	}
318
319	private Task MqttClientOnDisconnectedAsync(MqttClientDisconnectedEventArgs arg)
320	{
321	if (_disposed \|\| _stopping.IsCancellationRequested)	13!
322	{
NEW 323	return Task.CompletedTask;	×
324	}
325
326	if (_hasConnected)	13!
327	{
328	_logger.LogDebug("MQTT disconnected: {Reason}", BuildErrorResponse(arg));	13✔
329	}
330	else
331	{
NEW 332	_logger.LogTrace("MQTT disconnected before the initial connection completed: {Reason}", BuildErrorResponse(arg));	×
333	}
334
335	MarkConnectionLost();	13✔
336	return Task.CompletedTask;	13✔
337	}
338
339	private Task MqttClientOnConnectedAsync(MqttClientConnectedEventArgs arg)
340	{
341	_hasConnected = true;	10✔
342	_logger.LogDebug("MQTT connected: {ResultCode}", arg.ConnectResult.ResultCode);	10✔
343	MarkConnected();	10✔
344	return Task.CompletedTask;	10✔
345	}
346
347	private async Task MqttClientOnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
348	{
349	var handlers = ApplicationMessageReceivedAsync;	4✔
350	if (handlers is null)	4!
351	{
NEW 352	return;	×
353	}
354
355	foreach (Func<MqttApplicationMessageReceivedEventArgs, Task> handler in handlers.GetInvocationList())	16✔
356	{
357	await handler(arg).ConfigureAwait(false);	4✔
358	}
359	}	4✔
360
361	private static string BuildErrorResponse(MqttClientDisconnectedEventArgs arg)
362	{
363	var sb = new System.Text.StringBuilder();	13✔
364
365	sb.AppendLine(CultureInfo.InvariantCulture, $"{arg.Exception?.Message} ({arg.Reason})");	13✔
366	var ex = arg.Exception?.InnerException;	13✔
367	while (ex != null)	16✔
368	{
369	sb.AppendLine(ex.Message);	3✔
370	ex = ex.InnerException;	3✔
371	}
372
373	return sb.ToString();	13✔
374	}
375
376	private static TaskCompletionSource CreateConnectionSignal()
377	{
378	return new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);	33✔
379	}
380
381	private void MarkConnected()
382	{
383	lock (_connectionStateLock)	32✔
384	{
385	_connectedSignal.TrySetResult();	32✔
386	}	32✔
387	}	32✔
388
389	private void MarkConnectionLost()
390	{
391	lock (_connectionStateLock)	40✔
392	{
393	if (_connectedSignal.Task.IsCompleted)	40✔
394	{
395	_connectedSignal = CreateConnectionSignal();	12✔
396	}
397	}	40✔
398
399	MarkAllSubscriptionsPending();	40✔
400	}	40✔
401
402	private async Task WaitForConnectionAsync(CancellationToken cancellationToken)
403	{
404	while (!_mqttClient.IsConnected)	14✔
405	{
406	Task connectedTask;
407	lock (_connectionStateLock)	6✔
408	{
409	if (_mqttClient.IsConnected)	6!
410	{
NEW 411	return;	×
412	}
413
414	connectedTask = _connectedSignal.Task;	6✔
415	}	6✔
416
417	await connectedTask.WaitAsync(cancellationToken).ConfigureAwait(false);	6✔
418	}
419	}	8✔
420
421	private async Task DelayReconnectAsync(CancellationToken cancellationToken)
422	{
423	try
424	{
425	await Task.Delay(DefaultReconnectDelay, _timeProvider, cancellationToken).ConfigureAwait(false);	5✔
426	}	1✔
427	catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)	1✔
428	{
429	}	1✔
430	}	2✔
431
432	private string? GetConfiguredHost()
433	{
434	return (_clientOptions.ChannelOptions as MqttClientTcpOptions)?.RemoteEndpoint switch	27!
435	{	27✔
436	System.Net.DnsEndPoint dnsEndPoint => dnsEndPoint.Host,	27✔
NEW 437	System.Net.IPEndPoint ipEndPoint => ipEndPoint.Address.ToString(),	×
NEW 438	_ => null	×
439	};	27✔
440	}
441
442	private int? GetConfiguredPort()
443	{
444	return (_clientOptions.ChannelOptions as MqttClientTcpOptions)?.RemoteEndpoint switch	27!
445	{	27✔
446	System.Net.DnsEndPoint dnsEndPoint => dnsEndPoint.Port,	27✔
NEW 447	System.Net.IPEndPoint ipEndPoint => ipEndPoint.Port,	×
NEW 448	_ => null	×
449	};	27✔
450	}
451
452	/// <inheritdoc />
453	public void Dispose()
454	{
455	if (_disposed)	11!
456	{
UNCOV 457	return;	×
458	}
459
460	_disposed = true;	11✔
461	_logger.LogTrace("MQTT disconnecting");	11✔
462	_stopping.Cancel();	11✔
463	_publishQueue.Writer.TryComplete();	11✔
464
465	try
466	{
467	Task.WaitAll([_connectionTask, _publishTask], TimeSpan.FromSeconds(1));	11✔
NEW 468	}	×
469	catch (AggregateException)	11✔
470	{
471	}	11✔
472
473	_stopping.Dispose();	11✔
474	_clientOperationGate.Dispose();	11✔
475
476	if (_mqttClient is IDisposable disposable)	11!
477	{
478	disposable.Dispose();	11✔
479	}
480	}	11✔
481	}

net-daemon / netdaemon / 29408474270

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous