13468797469

Committed 19 Feb 2025 08:05PM UTC coverage: 84.158% (+0.5%) from 83.694%

Build # 13468797469

Build Type

push

github

Committed by

web-flow

Commit Message

Use MSBuild to load projects, instead of manual parsing (#1266)

For users (like me) who use a property to specify all their netdaemon versions, instead of individual versions, the existing checks are too simple. I've updated it to instead use msbuild to actually load the project file and determine the real version that nuget will see.

Run Details

844 of 1125 branches covered (75.02%)

Branch coverage included in aggregate %.

5 of 6 new or added lines in 2 files covered. (83.33%)

1 existing line in 1 file now uncovered.

3342 of 3849 relevant lines covered (86.83%)

313.98 hits per line

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

91.24

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

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 CancellationTokenSource _internalCancelSource = new();

    private readonly Subject<HassMessage> _hassMessageSubject = new();
    private readonly Task _handleNewMessagesTask;

    private const int WaitForResultTimeout = 20000;

    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>
    public HomeAssistantConnection(
        ILogger<IHomeAssistantConnection> logger,
        IWebSocketClientTransportPipeline pipeline,
        IHomeAssistantApiManager apiManager
    )
    {
        _transportPipeline = pipeline ?? throw new ArgumentNullException(nameof(pipeline));
        _apiManager = apiManager;
        _logger = logger;

        _resultMessageHandler = new ResultMessageHandler(_logger, TimeProvider.System);

        // 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(returnMessageTask, 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 resultMessageTask.WaitAsync(TimeSpan.FromMilliseconds(WaitForResultTimeout), cancelToken);

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

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

            // We make a task that subscribe for the return result message
            // this task will be returned and handled by caller
            var resultEvent = _hassMessageSubject
                .Where(n => n.Type == "result" && n.Id == command.Id)
                .FirstAsync().ToTask(CancellationToken.None);
            // We dont want to pass the incoming CancellationToken here because it will throw a TaskCanceledException
            // when calling services from an Apps Dispose(Async) and hide possible actual exceptions

            await _transportPipeline.SendMessageAsync(command, cancelToken);

            return resultEvent;
        }
        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)
                    {
                        _hassMessageSubject.OnNext(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");
            // make sure we always cancel any blocking operations
            if (!_internalCancelSource.IsCancellationRequested)
                await _internalCancelSource.CancelAsync();
        }
    }

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

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

net-daemon / netdaemon / 13468797469

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