26677597664

Committed 30 May 2026 07:04AM UTC coverage: 84.512% (+0.6%) from 83.944%

Build # 26677597664

Build Type

Pull #1385

github

Committed by

web-flow

Commit Message

Merge ca1957218 into 0116c9815

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

Coverage Stats

906 of 1217 branches covered (74.45%)

Branch coverage included in aggregate %.

113 of 135 new or added lines in 5 files covered. (83.7%)

9 existing lines in 4 files now uncovered.

3514 of 4013 relevant lines covered (87.57%)

138.44 hits per line

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

77.96

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

    /// <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.LogDebug("MQTT connection rejected: {ResultCode} {ReasonString}",
                        connectResult.ResultCode, connectResult.ReasonString);
                    await Task.Delay(ReconnectDelay, cancellationToken).ConfigureAwait(false);
                    continue;
                }

                _logger.LogTrace("MQTT client is ready");
                SignalConnection();
                await ResubscribeAsync(cancellationToken).ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                _logger.LogDebug(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)
    {
        _logger.LogDebug("MQTT disconnected: {Reason}", BuildErrorResponse(arg));
        return Task.CompletedTask;
    }

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

net-daemon / netdaemon / 26677597664

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