26678265425

Committed 30 May 2026 07:37AM UTC coverage: 84.739% (+0.8%) from 83.944%

Build # 26678265425

Build Type

Pull #1385

github

Committed by

web-flow

Commit Message

Merge b4372734e into 0116c9815

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

Coverage Stats

913 of 1223 branches covered (74.65%)

Branch coverage included in aggregate %.

122 of 142 new or added lines in 5 files covered. (85.92%)

2 existing lines in 2 files now uncovered.

3529 of 4019 relevant lines covered (87.81%)

409.27 hits per line

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

80.3

/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;

namespace NetDaemon.Extensions.MqttEntityManager;

/// <summary>
/// Wrapper to assure an MQTT connection.
/// </summary>
internal class AssuredMqttConnection : IAssuredMqttConnection, IDisposable
{
    private static readonly TimeSpan ReconnectDelay = TimeSpan.FromSeconds(5);
    private static readonly TimeSpan ConnectedPollDelay = TimeSpan.FromSeconds(1);

    private readonly ILogger<AssuredMqttConnection> _logger;
    private readonly IMqttClient _mqttClient;
    private readonly MqttClientOptions _clientOptions;
    private readonly Channel<MqttApplicationMessage> _publishQueue = Channel.CreateUnbounded<MqttApplicationMessage>(
        new UnboundedChannelOptions
        {
            SingleReader = true,
            SingleWriter = false
        });
    private readonly ConcurrentDictionary<string, MqttTopicFilter> _subscriptions = new();
    private readonly CancellationTokenSource _stopping = new();
    private readonly SemaphoreSlim _connectionSignal = new(0, 1);
    private readonly Task _connectionTask;
    private readonly Task _publishTask;
    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>
    public AssuredMqttConnection(
        ILogger<AssuredMqttConnection> logger,
        IMqttClientOptionsFactory mqttClientOptionsFactory,
        IMqttFactory mqttFactory,
        IOptions<MqttConfiguration> mqttConfig)
    {
        _logger = logger;

        _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;

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

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

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

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

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

                _logger.LogTrace("MQTT client is ready");
                _hasConnected = true;
                SignalConnection();
                await ResubscribeAsync(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))
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    if (!_mqttClient.IsConnected)
                    {
                        await WaitForConnectionAsync(cancellationToken).ConfigureAwait(false);
                        continue;
                    }

                    await _mqttClient.PublishAsync(message, cancellationToken).ConfigureAwait(false);
                    break;
                }
                catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
                {
                    break;
                }
                catch (Exception ex)
                {
                    _logger.LogDebug(ex, "Failed to publish MQTT message. The message will be retried after reconnect.");
                    await WaitForConnectionAsync(cancellationToken).ConfigureAwait(false);
                }
            }
        }
    }

    private async Task ResubscribeAsync(CancellationToken cancellationToken)
    {
        foreach (var topicFilter in _subscriptions.Values)
        {
            await SubscribeOnClientAsync(topicFilter, cancellationToken).ConfigureAwait(false);
        }
    }

    private async Task SubscribeOnClientAsync(MqttTopicFilter topicFilter, CancellationToken cancellationToken)
    {
        var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
            .WithTopicFilter(topicFilter)
            .Build();

        await _mqttClient.SubscribeAsync(subscribeOptions, cancellationToken).ConfigureAwait(false);
    }

    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));
        }

        return Task.CompletedTask;
    }

    private Task MqttClientOnConnectedAsync(MqttClientConnectedEventArgs arg)
    {
        _hasConnected = true;
        _logger.LogDebug("MQTT connected: {ResultCode}", arg.ConnectResult.ResultCode);
        SignalConnection();
        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 void SignalConnection()
    {
        if (_connectionSignal.CurrentCount == 0)
        {
            _connectionSignal.Release();
        }
    }

    private async Task WaitForConnectionAsync(CancellationToken cancellationToken)
    {
        if (_mqttClient.IsConnected)
        {
            return;
        }

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

    private static async Task DelayReconnectAsync(CancellationToken cancellationToken)
    {
        try
        {
            await Task.Delay(ReconnectDelay, 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();
        _connectionSignal.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
9	namespace NetDaemon.Extensions.MqttEntityManager;
10
11	/// <summary>
12	/// Wrapper to assure an MQTT connection.
13	/// </summary>
14	internal class AssuredMqttConnection : IAssuredMqttConnection, IDisposable
15	{
16	private static readonly TimeSpan ReconnectDelay = TimeSpan.FromSeconds(5);	3✔
17	private static readonly TimeSpan ConnectedPollDelay = TimeSpan.FromSeconds(1);	3✔
18
19	private readonly ILogger<AssuredMqttConnection> _logger;
20	private readonly IMqttClient _mqttClient;
21	private readonly MqttClientOptions _clientOptions;
22	private readonly Channel<MqttApplicationMessage> _publishQueue = Channel.CreateUnbounded<MqttApplicationMessage>(	14✔
23	new UnboundedChannelOptions	14✔
24	{	14✔
25	SingleReader = true,	14✔
26	SingleWriter = false	14✔
27	});	14✔
28	private readonly ConcurrentDictionary<string, MqttTopicFilter> _subscriptions = new();	14✔
29	private readonly CancellationTokenSource _stopping = new();	14✔
30	private readonly SemaphoreSlim _connectionSignal = new(0, 1);	14✔
31	private readonly Task _connectionTask;
32	private readonly Task _publishTask;
33	private bool _disposed;
34	private volatile bool _hasConnected;
35
36	/// <summary>
37	/// Initializes a new instance of the <see cref="AssuredMqttConnection"/> class.
38	/// </summary>
39	/// <param name="logger">The logger.</param>
40	/// <param name="mqttClientOptionsFactory">The MQTT client options factory.</param>
41	/// <param name="mqttFactory">The MQTT factory wrapper.</param>
42	/// <param name="mqttConfig">The MQTT configuration.</param>
43	public AssuredMqttConnection(	14✔
44	ILogger<AssuredMqttConnection> logger,	14✔
45	IMqttClientOptionsFactory mqttClientOptionsFactory,	14✔
46	IMqttFactory mqttFactory,	14✔
47	IOptions<MqttConfiguration> mqttConfig)	14✔
48	{
49	_logger = logger;	14✔
50
51	_logger.LogTrace("MQTT initiating connection");	14✔
52	_clientOptions = mqttClientOptionsFactory.CreateClientOptions(mqttConfig.Value);	14✔
53	_mqttClient = mqttFactory.CreateMqttClient();	14✔
54
55	_mqttClient.ConnectedAsync += MqttClientOnConnectedAsync;	14✔
56	_mqttClient.DisconnectedAsync += MqttClientOnDisconnectedAsync;	14✔
57	_mqttClient.ApplicationMessageReceivedAsync += MqttClientOnApplicationMessageReceivedAsync;	14✔
58
59	_connectionTask = Task.Run(() => MaintainConnectionAsync(_stopping.Token));	28✔
60	_publishTask = Task.Run(() => PublishQueuedMessagesAsync(_stopping.Token));	28✔
61	}	14✔
62
63	/// <inheritdoc />
64	public event Func<MqttApplicationMessageReceivedEventArgs, Task>? ApplicationMessageReceivedAsync;
65
66	/// <inheritdoc />
67	public async Task PublishAsync(MqttApplicationMessage message)
68	{
69	ObjectDisposedException.ThrowIf(_disposed, this);	27✔
70	await _publishQueue.Writer.WriteAsync(message, _stopping.Token).ConfigureAwait(false);	27✔
71	}	27✔
72
73	/// <inheritdoc />
74	public async Task SubscribeAsync(MqttTopicFilter topicFilter)
75	{
76	ObjectDisposedException.ThrowIf(_disposed, this);	12✔
77
78	if (string.IsNullOrEmpty(topicFilter.Topic))	12!
79	{
NEW 80	throw new ArgumentException("MQTT topic filter must specify a topic.", nameof(topicFilter));	×
81	}
82
83	_subscriptions[topicFilter.Topic] = topicFilter;	12✔
84
85	if (_mqttClient.IsConnected)	12✔
86	{
87	await SubscribeOnClientAsync(topicFilter, _stopping.Token).ConfigureAwait(false);	2✔
88	}
89	}	12✔
90
91	private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
92	{
93	while (!cancellationToken.IsCancellationRequested)	38✔
94	{
95	try
96	{
97	if (_mqttClient.IsConnected)	38✔
98	{
99	await Task.Delay(ConnectedPollDelay, cancellationToken).ConfigureAwait(false);	20✔
100	continue;	9✔
101	}
102
103	_logger.LogTrace("Connecting to MQTT broker at {Host}:{Port}/{UserName}",	18!
104	GetConfiguredHost(), GetConfiguredPort(), _clientOptions.Credentials?.GetUserName(_clientOptions));	18✔
105
106	var connectResult = await _mqttClient.ConnectAsync(_clientOptions, cancellationToken).ConfigureAwait(false);	18✔
107
108	if (connectResult.ResultCode != MqttClientConnectResultCode.Success)	14!
109	{
NEW 110	_logger.LogTrace("MQTT connection rejected: {ResultCode} {ReasonString}",	×
NEW 111	connectResult.ResultCode, connectResult.ReasonString);	×
NEW 112	await Task.Delay(ReconnectDelay, cancellationToken).ConfigureAwait(false);	×
NEW 113	continue;	×
114	}
115
116	_logger.LogTrace("MQTT client is ready");	14✔
117	_hasConnected = true;	14✔
118	SignalConnection();	14✔
119	await ResubscribeAsync(cancellationToken).ConfigureAwait(false);	14✔
120	}	14✔
121	catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)	4✔
122	{
123	break;	4✔
124	}
125	catch (Exception ex)	4✔
126	{
127	_logger.LogTrace(ex, "MQTT connection attempt failed");	4✔
128	await DelayReconnectAsync(cancellationToken).ConfigureAwait(false);	4✔
129	}
130	}
131	}	4✔
132
133	private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
134	{
135	await foreach (var message in _publishQueue.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))	82!
136	{
137	while (!cancellationToken.IsCancellationRequested)	36✔
138	{
139	try
140	{
141	if (!_mqttClient.IsConnected)	36✔
142	{
143	await WaitForConnectionAsync(cancellationToken).ConfigureAwait(false);	9✔
144	continue;	9✔
145	}
146
147	await _mqttClient.PublishAsync(message, cancellationToken).ConfigureAwait(false);	27✔
148	break;	27✔
149	}
NEW 150	catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)	×
151	{
NEW 152	break;	×
153	}
154	catch (Exception ex)
155	{
NEW 156	_logger.LogDebug(ex, "Failed to publish MQTT message. The message will be retried after reconnect.");	×
NEW 157	await WaitForConnectionAsync(cancellationToken).ConfigureAwait(false);	×
158	}
159	}
160	}	27✔
NEW 161	}	×
162
163	private async Task ResubscribeAsync(CancellationToken cancellationToken)
164	{
165	foreach (var topicFilter in _subscriptions.Values)	48✔
166	{
167	await SubscribeOnClientAsync(topicFilter, cancellationToken).ConfigureAwait(false);	10✔
168	}
169	}	14✔
170
171	private async Task SubscribeOnClientAsync(MqttTopicFilter topicFilter, CancellationToken cancellationToken)
172	{
173	var subscribeOptions = new MqttClientSubscribeOptionsBuilder()	12✔
174	.WithTopicFilter(topicFilter)	12✔
175	.Build();	12✔
176
177	await _mqttClient.SubscribeAsync(subscribeOptions, cancellationToken).ConfigureAwait(false);	12✔
178	}	12✔
179
180	private Task MqttClientOnDisconnectedAsync(MqttClientDisconnectedEventArgs arg)
181	{
182	if (_disposed \|\| _stopping.IsCancellationRequested)	13!
183	{
NEW 184	return Task.CompletedTask;	×
185	}
186
187	if (_hasConnected)	13!
188	{
189	_logger.LogDebug("MQTT disconnected: {Reason}", BuildErrorResponse(arg));	13✔
190	}
191	else
192	{
NEW 193	_logger.LogTrace("MQTT disconnected before the initial connection completed: {Reason}", BuildErrorResponse(arg));	×
194	}
195
196	return Task.CompletedTask;	13✔
197	}
198
199	private Task MqttClientOnConnectedAsync(MqttClientConnectedEventArgs arg)
200	{
201	_hasConnected = true;	10✔
202	_logger.LogDebug("MQTT connected: {ResultCode}", arg.ConnectResult.ResultCode);	10✔
203	SignalConnection();	10✔
204	return Task.CompletedTask;	10✔
205	}
206
207	private async Task MqttClientOnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
208	{
209	var handlers = ApplicationMessageReceivedAsync;	4✔
210	if (handlers is null)	4!
211	{
NEW 212	return;	×
213	}
214
215	foreach (Func<MqttApplicationMessageReceivedEventArgs, Task> handler in handlers.GetInvocationList())	16✔
216	{
217	await handler(arg).ConfigureAwait(false);	4✔
218	}
219	}	4✔
220
221	private static string BuildErrorResponse(MqttClientDisconnectedEventArgs arg)
222	{
223	var sb = new System.Text.StringBuilder();	13✔
224
225	sb.AppendLine(CultureInfo.InvariantCulture, $"{arg.Exception?.Message} ({arg.Reason})");	13✔
226	var ex = arg.Exception?.InnerException;	13✔
227	while (ex != null)	16✔
228	{
229	sb.AppendLine(ex.Message);	3✔
230	ex = ex.InnerException;	3✔
231	}
232
233	return sb.ToString();	13✔
234	}
235
236	private void SignalConnection()
237	{
238	if (_connectionSignal.CurrentCount == 0)	24✔
239	{
240	_connectionSignal.Release();	20✔
241	}
242	}	24✔
243
244	private async Task WaitForConnectionAsync(CancellationToken cancellationToken)
245	{
246	if (_mqttClient.IsConnected)	9✔
247	{
248	return;	1✔
249	}
250
251	await _connectionSignal.WaitAsync(cancellationToken).ConfigureAwait(false);	8✔
252	}	9✔
253
254	private static async Task DelayReconnectAsync(CancellationToken cancellationToken)
255	{
256	try
257	{
258	await Task.Delay(ReconnectDelay, cancellationToken).ConfigureAwait(false);	4✔
259	}	1✔
NEW 260	catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)	×
261	{
NEW 262	}	×
263	}	1✔
264
265	private string? GetConfiguredHost()
266	{
267	return (_clientOptions.ChannelOptions as MqttClientTcpOptions)?.RemoteEndpoint switch	18!
268	{	18✔
269	System.Net.DnsEndPoint dnsEndPoint => dnsEndPoint.Host,	18✔
NEW 270	System.Net.IPEndPoint ipEndPoint => ipEndPoint.Address.ToString(),	×
NEW 271	_ => null	×
272	};	18✔
273	}
274
275	private int? GetConfiguredPort()
276	{
277	return (_clientOptions.ChannelOptions as MqttClientTcpOptions)?.RemoteEndpoint switch	18!
278	{	18✔
279	System.Net.DnsEndPoint dnsEndPoint => dnsEndPoint.Port,	18✔
NEW 280	System.Net.IPEndPoint ipEndPoint => ipEndPoint.Port,	×
NEW 281	_ => null	×
282	};	18✔
283	}
284
285	/// <inheritdoc />
286	public void Dispose()
287	{
288	if (_disposed)	4!
289	{
UNCOV 290	return;	×
291	}
292
293	_disposed = true;	4✔
294	_logger.LogTrace("MQTT disconnecting");	4✔
295	_stopping.Cancel();	4✔
296	_publishQueue.Writer.TryComplete();	4✔
297
298	try
299	{
300	Task.WaitAll([_connectionTask, _publishTask], TimeSpan.FromSeconds(1));	4✔
NEW 301	}	×
302	catch (AggregateException)	4✔
303	{
304	}	4✔
305
306	_stopping.Dispose();	4✔
307	_connectionSignal.Dispose();	4✔
308
309	if (_mqttClient is IDisposable disposable)	4!
310	{
311	disposable.Dispose();	4✔
312	}
313	}	4✔
314	}

net-daemon / netdaemon / 26678265425

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