29100677284

Committed 10 Jul 2026 02:39PM UTC coverage: 83.43% (-0.2%) from 83.603%

Build # 29100677284

Build Type

push

github

Committed by

web-flow

Commit Message

Bump highbyte/sonarscan-dotnet from 2.5.0 to 2.5.1 (#1403)

Bumps [highbyte/sonarscan-dotnet](https://github.com/highbyte/sonarscan-dotnet) from 2.5.0 to 2.5.1.
- [Release notes](https://github.com/highbyte/sonarscan-dotnet/releases)
- [Commits](https://github.com/highbyte/sonarscan-dotnet/compare/v2.5.0...v2.5.1)

---
updated-dependencies:
- dependency-name: highbyte/sonarscan-dotnet
  dependency-version: 2.5.1
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Coverage Stats

899 of 1217 branches covered (73.87%)

Branch coverage included in aggregate %.

3431 of 3973 relevant lines covered (86.36%)

188.41 hits per line

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

93.72

/src/Client/NetDaemon.HassClient/Internal/HomeAssistantConnection.cs

using System.Collections.Concurrent;

namespace NetDaemon.Client.Internal;

internal class HomeAssistantConnection : IHomeAssistantConnection, IHomeAssistantHassMessages
{
    #region -- private declarations -

    private volatile bool _isDisposed;

    private readonly ILogger<IHomeAssistantConnection> _logger;
    private readonly IWebSocketClientTransportPipeline _transportPipeline;
    private readonly IHomeAssistantApiManager _apiManager;
    private readonly ResultMessageHandler _resultMessageHandler;
    private readonly TimeProvider _timeProvider;
    private readonly TimeSpan _waitForResultTimeout;
    private readonly CancellationTokenSource _internalCancelSource = new();

    private readonly Subject<HassMessage> _hassMessageSubject = new();
    private readonly ConcurrentDictionary<int, TaskCompletionSource<HassMessage>> _pendingResults = new();
    private readonly Task _handleNewMessagesTask;

    private static readonly TimeSpan DefaultWaitForResultTimeout = TimeSpan.FromSeconds(20);

    private readonly SemaphoreSlim _messageIdSemaphore = new(1, 1);
    private int _messageId = 1;
    private readonly AsyncLazy<IObservable<HassEvent>> _lazyAllEventsObservable;

    #endregion

    /// <summary>
    ///     Default constructor
    /// </summary>
    /// <param name="logger">A logger instance</param>
    /// <param name="pipeline">The pipeline to use for websocket communication</param>
    /// <param name="apiManager">The api manager</param>
    /// <param name="timeProvider">The time provider used when waiting for command results</param>
    /// <param name="waitForResultTimeout">The maximum time to wait for a command result</param>
    public HomeAssistantConnection(
        ILogger<IHomeAssistantConnection> logger,
        IWebSocketClientTransportPipeline pipeline,
        IHomeAssistantApiManager apiManager,
        TimeProvider? timeProvider = null,
        TimeSpan? waitForResultTimeout = null
    )
    {
        _transportPipeline = pipeline ?? throw new ArgumentNullException(nameof(pipeline));
        _apiManager = apiManager;
        _logger = logger;
        _timeProvider = timeProvider ?? TimeProvider.System;
        _waitForResultTimeout = waitForResultTimeout ?? DefaultWaitForResultTimeout;

        _resultMessageHandler = new ResultMessageHandler(_logger);

        // We lazily cache same observable for all events. There are no reason we should use multiple subscriptions
        // to all events. If people wants that they can provide a "*" type and get the same thing
        _lazyAllEventsObservable = new AsyncLazy<IObservable<HassEvent>>(async Task<IObservable<HassEvent>>() =>
            await SubscribeToHomeAssistantEventsInternalAsync(null, _internalCancelSource.Token));

        if (_transportPipeline.WebSocketState != WebSocketState.Open)
            throw new ApplicationException(
                $"Expected WebSocket state 'Open' got '{_transportPipeline.WebSocketState}'");

        _handleNewMessagesTask = Task.Factory.StartNew(async () => await HandleNewMessages().ConfigureAwait(false),
            TaskCreationOptions.LongRunning);
    }

    public async Task<IObservable<HassEvent>> SubscribeToHomeAssistantEventsAsync(string? eventType,
        CancellationToken cancelToken)
    {
        // When subscribe all events, optimize using the same IObservable<HassEvent> if we subscribe multiple times
        if (string.IsNullOrEmpty(eventType))
            return await _lazyAllEventsObservable.Value;

        return await SubscribeToHomeAssistantEventsInternalAsync(eventType, cancelToken);
    }

    private async Task<IObservable<HassEvent>> SubscribeToHomeAssistantEventsInternalAsync(string? eventType,
        CancellationToken cancelToken)
    {
        var combinedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(
            cancelToken,
            _internalCancelSource.Token
        );

        var result = await SendCommandAndReturnHassMessageResponseAsync(new SubscribeEventCommand(),
            combinedTokenSource.Token).ConfigureAwait(false);

        // The id if the message we used to subscribe should be used as the filter for the event messages
        var observableResult = _hassMessageSubject.Where(n => n.Type == "event" && n.Id == result?.Id)
            .Select(n => n.Event!);

        return observableResult;
    }

    public async Task WaitForConnectionToCloseAsync(CancellationToken cancelToken)
    {
        var combinedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(
            cancelToken,
            _internalCancelSource.Token
        );

        // Just wait for token source (internal och provided one)
        await combinedTokenSource.Token.AsTask().ConfigureAwait(false);
    }

    public async Task SendCommandAsync<T>(T command, CancellationToken cancelToken) where T : CommandMessage
    {
        var returnMessageTask = await SendCommandAsyncInternal(command, cancelToken);
        _resultMessageHandler.HandleResult(
            WaitForPendingResultAsync(command, returnMessageTask, _internalCancelSource.Token),
            command);
    }

    public async Task<TResult?> SendCommandAndReturnResponseAsync<T, TResult>(T command, CancellationToken cancelToken)
        where T : CommandMessage
    {
        var result = await SendCommandAndReturnResponseRawAsync(command, cancelToken).ConfigureAwait(false);

        return result is not null ? result.Value.Deserialize<TResult>() : default;
    }

    public async Task<JsonElement?> SendCommandAndReturnResponseRawAsync<T>(T command, CancellationToken cancelToken)
        where T : CommandMessage
    {
        var hassMessage =
            await SendCommandAndReturnHassMessageResponseAsync(command, cancelToken).ConfigureAwait(false);

        // The SendCommandsAndReturnHAssMessageResponse will throw if not successful so just ignore errors here
        return hassMessage?.ResultElement;
    }

    public async Task<HassMessage?> SendCommandAndReturnHassMessageResponseAsync<T>(T command,
        CancellationToken cancelToken)
        where T : CommandMessage
    {
        var resultMessageTask = await SendCommandAsyncInternal(command, cancelToken);
        var result = await WaitForPendingResultAsync(command, resultMessageTask, cancelToken).ConfigureAwait(false);

        if (result.Success ?? false)
            return result;

        // Non successful command should throw exception
        throw new InvalidOperationException(
            $"Failed command ({command.Type}) error: {result.Error}.  Sent command is {command.ToJsonElement()}");
    }

    public async ValueTask DisposeAsync()
    {
        if (_isDisposed) return;
        _isDisposed = true;

        try
        {
            await CloseAsync().ConfigureAwait(false);
        }
        catch (Exception e)
        {
            _logger.LogDebug(e, "Failed to close HomeAssistantConnection");
        }

        if (!_internalCancelSource.IsCancellationRequested)
            await _internalCancelSource.CancelAsync();
        CancelPendingResults();

        // Gracefully wait for task or timeout
        await Task.WhenAny(
            _handleNewMessagesTask,
            Task.Delay(5000)
        ).ConfigureAwait(false);

        await _transportPipeline.DisposeAsync().ConfigureAwait(false);
        _internalCancelSource.Dispose();
        _hassMessageSubject.Dispose();
        await _resultMessageHandler.DisposeAsync();
        _messageIdSemaphore.Dispose();
    }

    public Task<T?> GetApiCallAsync<T>(string apiPath, CancellationToken cancelToken)
    {
        ObjectDisposedException.ThrowIf(_isDisposed, this);
        return _apiManager.GetApiCallAsync<T>(apiPath, cancelToken);
    }

    public Task<T?> PostApiCallAsync<T>(string apiPath, CancellationToken cancelToken, object? data = null)
    {
        ObjectDisposedException.ThrowIf(_isDisposed, this);
        return _apiManager.PostApiCallAsync<T>(apiPath, cancelToken, data);
    }

    public IObservable<HassMessage> OnHassMessage => _hassMessageSubject;

    private async Task<Task<HassMessage>> SendCommandAsyncInternal<T>(T command, CancellationToken cancelToken) where T : CommandMessage
    {
        ObjectDisposedException.ThrowIf(_isDisposed, this);

        // The semaphore can fail to be taken in rare cases so we need
        // to keep this out of the try/finally block so it will not be released
        await _messageIdSemaphore.WaitAsync(cancelToken).ConfigureAwait(false);
        try
        {
            // We need to make sure messages to HA are send with increasing Ids therefore we need to synchronize
            // increasing the messageId and Sending the message
            command.Id = ++_messageId;

            var resultEvent = new TaskCompletionSource<HassMessage>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            if (!_pendingResults.TryAdd(command.Id, resultEvent))
                throw new InvalidOperationException($"A command with id {command.Id} is already pending");

            try
            {
                await _transportPipeline.SendMessageAsync(command, cancelToken);
            }
            catch (Exception)
            {
                _pendingResults.TryRemove(command.Id, out _);
                throw;
            }

            return resultEvent.Task;
        }
        finally
        {
            _messageIdSemaphore.Release();
        }
    }

    private async Task HandleNewMessages()
    {
        try
        {
            while (!_internalCancelSource.IsCancellationRequested)
            {
                var msg = await _transportPipeline.GetNextMessagesAsync<HassMessage>(_internalCancelSource.Token)
                    .ConfigureAwait(false);
                try
                {
                    foreach (var obj in msg)
                    {
                        HandleIncomingMessage(obj);
                    }
                }
                catch (Exception e)
                {
                    _logger.LogError(e, "Failed processing new message from Home Assistant");
                }
            }
        }
        catch (OperationCanceledException)
        {
            // Normal case just exit
        }
        finally
        {
            _logger.LogTrace("Stop processing new messages");
            CancelPendingResults();
            // make sure we always cancel any blocking operations
            if (!_internalCancelSource.IsCancellationRequested)
                await _internalCancelSource.CancelAsync();
        }
    }

    private async Task<HassMessage> WaitForPendingResultAsync(CommandMessage command,
        Task<HassMessage> resultMessageTask,
        CancellationToken cancelToken)
    {
        try
        {
            return await resultMessageTask.WaitAsync(_waitForResultTimeout, _timeProvider, cancelToken)
                .ConfigureAwait(false);
        }
        catch (TimeoutException)
        {
            CancelPendingResult(command.Id, CancellationToken.None);
            throw;
        }
        catch (OperationCanceledException)
        {
            CancelPendingResult(command.Id, cancelToken);
            throw;
        }
    }

    private void HandleIncomingMessage(HassMessage message)
    {
        if (message.Type == "result" && _pendingResults.TryRemove(message.Id, out var completionSource))
            completionSource.TrySetResult(message);

        _hassMessageSubject.OnNext(message);
    }

    private void CancelPendingResult(int messageId, CancellationToken cancelToken = default)
    {
        if (!_pendingResults.TryRemove(messageId, out var completionSource))
            return;

        if (cancelToken.CanBeCanceled)
            completionSource.TrySetCanceled(cancelToken);
        else
            completionSource.TrySetCanceled(CancellationToken.None);
    }

    private void CancelPendingResults()
    {
        foreach (var pendingResult in _pendingResults)
        {
            if (_pendingResults.TryRemove(pendingResult.Key, out var completionSource))
                completionSource.TrySetCanceled();
        }
    }

    private Task CloseAsync()
    {
        return _transportPipeline.CloseAsync();
    }
}

1	using System.Collections.Concurrent;
2
3	namespace NetDaemon.Client.Internal;
4
5	internal class HomeAssistantConnection : IHomeAssistantConnection, IHomeAssistantHassMessages
6	{
7	#region -- private declarations -
8
9	private volatile bool _isDisposed;
10
11	private readonly ILogger<IHomeAssistantConnection> _logger;
12	private readonly IWebSocketClientTransportPipeline _transportPipeline;
13	private readonly IHomeAssistantApiManager _apiManager;
14	private readonly ResultMessageHandler _resultMessageHandler;
15	private readonly TimeProvider _timeProvider;
16	private readonly TimeSpan _waitForResultTimeout;
17	private readonly CancellationTokenSource _internalCancelSource = new();	45✔
18
19	private readonly Subject<HassMessage> _hassMessageSubject = new();	45✔
20	private readonly ConcurrentDictionary<int, TaskCompletionSource<HassMessage>> _pendingResults = new();	45✔
21	private readonly Task _handleNewMessagesTask;
22
23	private static readonly TimeSpan DefaultWaitForResultTimeout = TimeSpan.FromSeconds(20);	3✔
24
25	private readonly SemaphoreSlim _messageIdSemaphore = new(1, 1);	45✔
26	private int _messageId = 1;	45✔
27	private readonly AsyncLazy<IObservable<HassEvent>> _lazyAllEventsObservable;
28
29	#endregion
30
31	/// <summary>
32	/// Default constructor
33	/// </summary>
34	/// <param name="logger">A logger instance</param>
35	/// <param name="pipeline">The pipeline to use for websocket communication</param>
36	/// <param name="apiManager">The api manager</param>
37	/// <param name="timeProvider">The time provider used when waiting for command results</param>
38	/// <param name="waitForResultTimeout">The maximum time to wait for a command result</param>
39	public HomeAssistantConnection(	45✔
40	ILogger<IHomeAssistantConnection> logger,	45✔
41	IWebSocketClientTransportPipeline pipeline,	45✔
42	IHomeAssistantApiManager apiManager,	45✔
43	TimeProvider? timeProvider = null,	45✔
44	TimeSpan? waitForResultTimeout = null	45✔
45	)	45✔
46	{
47	_transportPipeline = pipeline ?? throw new ArgumentNullException(nameof(pipeline));	45!
48	_apiManager = apiManager;	45✔
49	_logger = logger;	45✔
50	_timeProvider = timeProvider ?? TimeProvider.System;	45✔
51	_waitForResultTimeout = waitForResultTimeout ?? DefaultWaitForResultTimeout;	45✔
52
53	_resultMessageHandler = new ResultMessageHandler(_logger);	45✔
54
55	// We lazily cache same observable for all events. There are no reason we should use multiple subscriptions
56	// to all events. If people wants that they can provide a "*" type and get the same thing
57	_lazyAllEventsObservable = new AsyncLazy<IObservable<HassEvent>>(async Task<IObservable<HassEvent>>() =>	45✔
58	await SubscribeToHomeAssistantEventsInternalAsync(null, _internalCancelSource.Token));	56✔
59
60	if (_transportPipeline.WebSocketState != WebSocketState.Open)	45✔
61	throw new ApplicationException(	1✔
62	$"Expected WebSocket state 'Open' got '{_transportPipeline.WebSocketState}'");	1✔
63
64	_handleNewMessagesTask = Task.Factory.StartNew(async () => await HandleNewMessages().ConfigureAwait(false),	88✔
65	TaskCreationOptions.LongRunning);	44✔
66	}	44✔
67
68	public async Task<IObservable<HassEvent>> SubscribeToHomeAssistantEventsAsync(string? eventType,
69	CancellationToken cancelToken)
70	{
71	// When subscribe all events, optimize using the same IObservable<HassEvent> if we subscribe multiple times
72	if (string.IsNullOrEmpty(eventType))	31✔
73	return await _lazyAllEventsObservable.Value;	28✔
74
75	return await SubscribeToHomeAssistantEventsInternalAsync(eventType, cancelToken);	3✔
76	}	31✔
77
78	private async Task<IObservable<HassEvent>> SubscribeToHomeAssistantEventsInternalAsync(string? eventType,
79	CancellationToken cancelToken)
80	{
81	var combinedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(	14✔
82	cancelToken,	14✔
83	_internalCancelSource.Token	14✔
84	);	14✔
85
86	var result = await SendCommandAndReturnHassMessageResponseAsync(new SubscribeEventCommand(),	14✔
87	combinedTokenSource.Token).ConfigureAwait(false);	14✔
88
89	// The id if the message we used to subscribe should be used as the filter for the event messages
90	var observableResult = _hassMessageSubject.Where(n => n.Type == "event" && n.Id == result?.Id)	1,491!
91	.Select(n => n.Event!);	924✔
92
93	return observableResult;	14✔
94	}	14✔
95
96	public async Task WaitForConnectionToCloseAsync(CancellationToken cancelToken)
97	{
98	var combinedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(	12✔
99	cancelToken,	12✔
100	_internalCancelSource.Token	12✔
101	);	12✔
102
103	// Just wait for token source (internal och provided one)
104	await combinedTokenSource.Token.AsTask().ConfigureAwait(false);	12✔
105	}	×
106
107	public async Task SendCommandAsync<T>(T command, CancellationToken cancelToken) where T : CommandMessage
108	{
109	var returnMessageTask = await SendCommandAsyncInternal(command, cancelToken);	19✔
110	_resultMessageHandler.HandleResult(	18✔
111	WaitForPendingResultAsync(command, returnMessageTask, _internalCancelSource.Token),	18✔
112	command);	18✔
113	}	18✔
114
115	public async Task<TResult?> SendCommandAndReturnResponseAsync<T, TResult>(T command, CancellationToken cancelToken)
116	where T : CommandMessage
117	{
118	var result = await SendCommandAndReturnResponseRawAsync(command, cancelToken).ConfigureAwait(false);	118✔
119
120	return result is not null ? result.Value.Deserialize<TResult>() : default;	117✔
121	}	117✔
122
123	public async Task<JsonElement?> SendCommandAndReturnResponseRawAsync<T>(T command, CancellationToken cancelToken)
124	where T : CommandMessage
125	{
126	var hassMessage =	123✔
127	await SendCommandAndReturnHassMessageResponseAsync(command, cancelToken).ConfigureAwait(false);	123✔
128
129	// The SendCommandsAndReturnHAssMessageResponse will throw if not successful so just ignore errors here
130	return hassMessage?.ResultElement;	122!
131	}	122✔
132
133	public async Task<HassMessage?> SendCommandAndReturnHassMessageResponseAsync<T>(T command,
134	CancellationToken cancelToken)
135	where T : CommandMessage
136	{
137	var resultMessageTask = await SendCommandAsyncInternal(command, cancelToken);	150✔
138	var result = await WaitForPendingResultAsync(command, resultMessageTask, cancelToken).ConfigureAwait(false);	148✔
139
140	if (result.Success ?? false)	145✔
141	return result;	143✔
142
143	// Non successful command should throw exception
144	throw new InvalidOperationException(	2✔
145	$"Failed command ({command.Type}) error: {result.Error}. Sent command is {command.ToJsonElement()}");	2✔
146	}	143✔
147
148	public async ValueTask DisposeAsync()
149	{
150	if (_isDisposed) return;	46✔
151	_isDisposed = true;	44✔
152
153	try
154	{
155	await CloseAsync().ConfigureAwait(false);	44✔
156	}	44✔
157	catch (Exception e)	×
158	{
159	_logger.LogDebug(e, "Failed to close HomeAssistantConnection");	×
160	}	×
161
162	if (!_internalCancelSource.IsCancellationRequested)	44✔
163	await _internalCancelSource.CancelAsync();	30✔
164	CancelPendingResults();	44✔
165
166	// Gracefully wait for task or timeout
167	await Task.WhenAny(	44✔
168	_handleNewMessagesTask,	44✔
169	Task.Delay(5000)	44✔
170	).ConfigureAwait(false);	44✔
171
172	await _transportPipeline.DisposeAsync().ConfigureAwait(false);	44✔
173	_internalCancelSource.Dispose();	44✔
174	_hassMessageSubject.Dispose();	44✔
175	await _resultMessageHandler.DisposeAsync();	44✔
176	_messageIdSemaphore.Dispose();	44✔
177	}	45✔
178
179	public Task<T?> GetApiCallAsync<T>(string apiPath, CancellationToken cancelToken)
180	{
181	ObjectDisposedException.ThrowIf(_isDisposed, this);	34✔
182	return _apiManager.GetApiCallAsync<T>(apiPath, cancelToken);	33✔
183	}
184
185	public Task<T?> PostApiCallAsync<T>(string apiPath, CancellationToken cancelToken, object? data = null)
186	{
187	ObjectDisposedException.ThrowIf(_isDisposed, this);	2✔
188	return _apiManager.PostApiCallAsync<T>(apiPath, cancelToken, data);	1✔
189	}
190
191	public IObservable<HassMessage> OnHassMessage => _hassMessageSubject;	4✔
192
193	private async Task<Task<HassMessage>> SendCommandAsyncInternal<T>(T command, CancellationToken cancelToken) where T : CommandMessage
194	{
195	ObjectDisposedException.ThrowIf(_isDisposed, this);	169✔
196
197	// The semaphore can fail to be taken in rare cases so we need
198	// to keep this out of the try/finally block so it will not be released
199	await _messageIdSemaphore.WaitAsync(cancelToken).ConfigureAwait(false);	168✔
200	try
201	{
202	// We need to make sure messages to HA are send with increasing Ids therefore we need to synchronize
203	// increasing the messageId and Sending the message
204	command.Id = ++_messageId;	168✔
205
206	var resultEvent = new TaskCompletionSource<HassMessage>(	168✔
207	TaskCreationOptions.RunContinuationsAsynchronously);	168✔
208	if (!_pendingResults.TryAdd(command.Id, resultEvent))	168!
209	throw new InvalidOperationException($"A command with id {command.Id} is already pending");	×
210
211	try
212	{
213	await _transportPipeline.SendMessageAsync(command, cancelToken);	168✔
214	}	166✔
215	catch (Exception)	2✔
216	{
217	_pendingResults.TryRemove(command.Id, out _);	2✔
218	throw;	2✔
219	}
220
221	return resultEvent.Task;	166✔
222	}
223	finally
224	{
225	_messageIdSemaphore.Release();	168✔
226	}
227	}	166✔
228
229	private async Task HandleNewMessages()
230	{
231	try
232	{
233	while (!_internalCancelSource.IsCancellationRequested)	250✔
234	{
235	var msg = await _transportPipeline.GetNextMessagesAsync<HassMessage>(_internalCancelSource.Token)	248✔
236	.ConfigureAwait(false);	248✔
237	try
238	{
239	foreach (var obj in msg)	1,016✔
240	{
241	HandleIncomingMessage(obj);	302✔
242	}
243	}	206✔
244	catch (Exception e)	×
245	{
246	_logger.LogError(e, "Failed processing new message from Home Assistant");	×
247	}	×
248	}
249	}	2✔
250	catch (OperationCanceledException)	8✔
251	{
252	// Normal case just exit
253	}	8✔
254	finally
255	{
256	_logger.LogTrace("Stop processing new messages");	32✔
257	CancelPendingResults();	32✔
258	// make sure we always cancel any blocking operations
259	if (!_internalCancelSource.IsCancellationRequested)	32✔
260	await _internalCancelSource.CancelAsync();	14✔
261	}
262	}	10✔
263
264	private async Task<HassMessage> WaitForPendingResultAsync(CommandMessage command,
265	Task<HassMessage> resultMessageTask,
266	CancellationToken cancelToken)
267	{
268	try
269	{
270	return await resultMessageTask.WaitAsync(_waitForResultTimeout, _timeProvider, cancelToken)	166✔
271	.ConfigureAwait(false);	166✔
272	}
273	catch (TimeoutException)	1✔
274	{
275	CancelPendingResult(command.Id, CancellationToken.None);	1✔
276	throw;	1✔
277	}
278	catch (OperationCanceledException)	4✔
279	{
280	CancelPendingResult(command.Id, cancelToken);	4✔
281	throw;	4✔
282	}
283	}	161✔
284
285	private void HandleIncomingMessage(HassMessage message)
286	{
287	if (message.Type == "result" && _pendingResults.TryRemove(message.Id, out var completionSource))	302✔
288	completionSource.TrySetResult(message);	161✔
289
290	_hassMessageSubject.OnNext(message);	302✔
291	}	302✔
292
293	private void CancelPendingResult(int messageId, CancellationToken cancelToken = default)
294	{
295	if (!_pendingResults.TryRemove(messageId, out var completionSource))	5✔
296	return;	1✔
297
298	if (cancelToken.CanBeCanceled)	4✔
299	completionSource.TrySetCanceled(cancelToken);	3✔
300	else
301	completionSource.TrySetCanceled(CancellationToken.None);	1✔
302	}	1✔
303
304	private void CancelPendingResults()
305	{
306	foreach (var pendingResult in _pendingResults)	154✔
307	{
308	if (_pendingResults.TryRemove(pendingResult.Key, out var completionSource))	1✔
309	completionSource.TrySetCanceled();	1✔
310	}
311	}	76✔
312
313	private Task CloseAsync()
314	{
315	return _transportPipeline.CloseAsync();	44✔
316	}
317	}

net-daemon / netdaemon / 29100677284

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