Enable Apps to call services when app shuts down in Dispose(Async)() and OnComplete of HA events (#1140)

* Enable Apps to call services when app shuts down in Dispose(Async)() and OnCOmplete of HA events

* Fix warninng

Author: FrankBakkerNl

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

@@ -164,21 +164,21 @@ await Task.WhenAny(
 
     public Task<T?> GetApiCallAsync<T>(string apiPath, CancellationToken cancelToken)
     {
-        ObjectDisposedException.ThrowIf(_isDisposed, nameof(HomeAssistantConnection));
+        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, nameof(HomeAssistantConnection));
+        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, nameof(HomeAssistantConnection));
+        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

src/HassModel/NetDaemon.HassModel.Tests/Internal/AppScopedHaContextProviderTest.cs

@@ -280,6 +280,27 @@ public async Task EventsStopAfterDispose()
         eventObserverMock.VerifyNoOtherCalls();
     }
 
+    [Fact]
+    public async Task CanCallServiceInEventCompleteDueToDispose()
+    {
+        // Arrange
+        var provider = await CreateServiceProvider();
+        var serviceScope = provider.CreateScope();
+        var haContext = serviceScope.ServiceProvider.GetRequiredService<IHaContext>();
+
+        haContext.Events.Subscribe(onNext: _ => { }, onError: _ => { }, onCompleted:()=>{haContext.CallService("Light", "turn_on");});
+
+        // Disposing the entire Scope will Dispose the AppScopedHaContextProvider, which will then first Dispose its QueuedObservable.
+        // That triggers haContext.Events.OnComplete().
+        // We want to make sure that any code in event handlers for OnComplete is still able to call services
+        await ((IAsyncDisposable)serviceScope).DisposeAsync();
+
+        // Assert
+        _hassConnectionMock.Verify(
+            c => c.SendCommandAsync(It.IsAny<CallServiceCommand>(),
+                It.IsAny<CancellationToken>()), Times.Once);
+    }
+
     [Fact]
     public async Task TestThatCallServiceTrackBackgroundTask()
     {
@@ -368,6 +389,5 @@ public record TestEventData(string command, int endpoint_id, string otherField);
     public void Dispose()
     {
         _hassEventSubjectMock.Dispose();
-        GC.SuppressFinalize(this);
     }
 }

src/HassModel/NetDaemon.HassModel.Tests/Internal/QueuedObservabeTest.cs

@@ -1,19 +1,19 @@
 using System.Reactive.Subjects;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
 using NetDaemon.HassModel.Internal;
 using NetDaemon.HassModel.Tests.TestHelpers;
 
 namespace NetDaemon.HassModel.Tests.Internal;
 
-public class QueuedObservabeTest
+public class QueuedObservableTest
 {
     [Fact]
-    public async Task EventsSouldbeforwarded()
+    public async Task OnNextShouldBeForwarded()
     {
         var source = new Subject<int>();
 
-        var queue = new QueuedObservable<int>(Mock.Of<ILogger<IHaContext>>());
-        queue.Initialize(source);
+        var queue = new QueuedObservable<int>(source, NullLogger.Instance);
 
         var subscriber = new Mock<IObserver<int>>();
         queue.Subscribe(subscriber.Object);
@@ -27,20 +27,48 @@ public async Task EventsSouldbeforwarded()
         subscriber.Verify(s => s.OnNext(3), Times.Once);
     }
 
+    [Fact]
+    public async Task OnErrorShouldBeForwarded()
+    {
+        var source = new Subject<int>();
+        var queue = new QueuedObservable<int>(source, NullLogger.Instance);
+        var subscriber = new Mock<IObserver<int>>();
+        queue.Subscribe(subscriber.Object);
+
+        var exception = new Exception("TestException");
+        source.OnError(exception);
+
+        await queue.DisposeAsync();
+        subscriber.Verify(s => s.OnError(exception));
+    }
+
+    [Fact]
+    public async Task? OnCompletedShouldBeForwarded()
+    {
+        var source = new Subject<int>();
+        var queue = new QueuedObservable<int>(source, NullLogger.Instance);
+        var subscriber = new Mock<IObserver<int>>();
+
+        queue.Subscribe(subscriber.Object);
+
+        source.OnCompleted();
+
+        await queue.DisposeAsync();
+        subscriber.Verify(s => s.OnCompleted());
+    }
+
     [Fact]
     public async Task SubscribersShouldNotBlockEachOther()
     {
         var source = new Subject<int>();
 
-        var queue1 = new QueuedObservable<int>(Mock.Of<ILogger<IHaContext>>());
-        queue1.Initialize(source);
+        var queue1 = new QueuedObservable<int>(source, NullLogger.Instance);
 
         var subscriber = new Mock<IObserver<int>>();
         queue1.Subscribe(subscriber.Object);
 
         // The second subscriber will block the first event
-        var queue2 = new QueuedObservable<int>(Mock.Of<ILogger<IHaContext>>());
-        queue2.Initialize(source);
+        var queue2 = new QueuedObservable<int>(source, NullLogger.Instance);
         var blockSubscribers = new ManualResetEvent(false);
         queue2.Subscribe(_ => blockSubscribers.WaitOne());
 
@@ -62,12 +90,9 @@ public async Task SubscribersShouldNotBlockEachOther()
     public async Task WhenScopeIsDisposedSubscribersAreDetached()
     {
         var testSubject = new Subject<string>();
-        var loggerMock = new Mock<ILogger<IHaContext>>();
         // Create 2 ScopedObservables for the same subject
-        var scoped1 = new QueuedObservable<string>(loggerMock.Object);
-        scoped1.Initialize(testSubject);
-        var scoped2 = new QueuedObservable<string>(loggerMock.Object);
-        scoped2.Initialize(testSubject);
+        var scoped1 = new QueuedObservable<string>(testSubject, NullLogger.Instance);
+        var scoped2 = new QueuedObservable<string>(testSubject, NullLogger.Instance);
 
         // First scope has 2 subscribers, second has 1
         var scope1AObserverMock = new Mock<IObserver<string>>();
@@ -78,11 +103,11 @@ public async Task WhenScopeIsDisposedSubscribersAreDetached()
         var scope2ObserverMock = new Mock<IObserver<string>>();
         scoped2.Subscribe(scope2ObserverMock.Object);
 
-        var waitTasks = new Task[]
+        var waitTasks = new[]
         {
-                scope1AObserverMock.WaitForInvocationAndVerify(o => o.OnNext("Event1")),
-                scope1BObserverMock.WaitForInvocationAndVerify(o => o.OnNext("Event1")),
-                scope2ObserverMock.WaitForInvocationAndVerify(o => o.OnNext("Event1"))
+            scope1AObserverMock.WaitForInvocationAndVerify(o => o.OnNext("Event1")),
+            scope1BObserverMock.WaitForInvocationAndVerify(o => o.OnNext("Event1")),
+            scope2ObserverMock.WaitForInvocationAndVerify(o => o.OnNext("Event1"))
         };
 
         // Now start firing events
@@ -109,8 +134,7 @@ public async Task WhenScopeIsDisposedSubscribersAreDetached()
     public async Task TestQueuedObservableShouldHaveFinishedTasksOnDispose()
     {
         var source = new Subject<int>();
-        var queue = new QueuedObservable<int>(Mock.Of<ILogger<IHaContext>>());
-        queue.Initialize(source);
+        var queue = new QueuedObservable<int>(source, NullLogger.Instance);
         var subscriber = new Mock<IObserver<int>>();
         queue.Subscribe(subscriber.Object);
 
@@ -129,8 +153,7 @@ public async Task TestQueuedObservableShouldLogOnException()
     {
         var source = new Subject<int>();
         var loggerMock = new Mock<ILogger<IHaContext>>();
-        var queue = new QueuedObservable<int>(loggerMock.Object);
-        queue.Initialize(source);
+        var queue = new QueuedObservable<int>(source,loggerMock.Object);
         var subscriber = new Mock<IObserver<int>>();
         subscriber.Setup(n => n.OnNext(1)).Throws<InvalidOperationException>();
         queue.Subscribe(subscriber.Object);
@@ -152,9 +175,7 @@ public async Task TestQueuedObservableShouldLogOnException()
     public async Task TestQueuedObservableShouldStillHaveSubscribersOnException()
     {
         var source = new Subject<int>();
-        var loggerMock = new Mock<ILogger<IHaContext>>();
-        var queue = new QueuedObservable<int>(loggerMock.Object);
-        queue.Initialize(source);
+        var queue = new QueuedObservable<int>(source, NullLogger.Instance);
         var subscriber = new Mock<IObserver<int>>();
         subscriber.Setup(n => n.OnNext(1)).Throws<InvalidOperationException>();
         queue.Subscribe(subscriber.Object);

src/HassModel/NetDeamon.HassModel/DependencyInjectionSetup.cs

@@ -1,5 +1,4 @@
 using Microsoft.Extensions.Hosting;
-using NetDaemon.Infrastructure.ObservableHelpers;
 
 namespace NetDaemon.HassModel;
 
@@ -36,8 +35,6 @@ public static void AddScopedHaContext(this IServiceCollection services)
         services.AddScoped<IBackgroundTaskTracker>(s => s.GetRequiredService<BackgroundTaskTracker>());
         services.AddTransient<ICacheManager, CacheManager>();
         services.AddTransient<IHaContext>(s => s.GetRequiredService<AppScopedHaContextProvider>());
-        services.AddScoped<QueuedObservable<HassEvent>>();
-        services.AddScoped<IQueuedObservable<HassEvent>>(s => s.GetRequiredService<QueuedObservable<HassEvent>>());
         services.AddScoped<TriggerManager>();
         services.AddTransient<ITriggerManager>(s => s.GetRequiredService<TriggerManager>());
     }

src/HassModel/NetDeamon.HassModel/IQueuedObservable.cs

@@ -1,5 +1,4 @@
 using NetDaemon.Client.HomeAssistant.Extensions;
-using NetDaemon.Infrastructure.ObservableHelpers;
 
 namespace NetDaemon.HassModel.Internal;
 
@@ -11,34 +10,34 @@ namespace NetDaemon.HassModel.Internal;
 internal class AppScopedHaContextProvider : IHaContext, IAsyncDisposable
 {
     private volatile bool _isDisposed;
+    private volatile bool _isDisposing;
     private readonly IHomeAssistantApiManager _apiManager;
     private readonly EntityStateCache _entityStateCache;
 
     private readonly IHomeAssistantRunner _hassRunner;
-    private readonly IQueuedObservable<HassEvent> _queuedObservable;
+    private readonly QueuedObservable<HassEvent> _queuedObservable;
     private readonly IBackgroundTaskTracker _backgroundTaskTracker;
+    private readonly ILogger<AppScopedHaContextProvider> _logger;
 
     private readonly CancellationTokenSource _tokenSource = new();
 
     public AppScopedHaContextProvider(
         EntityStateCache entityStateCache,
         IHomeAssistantRunner hassRunner,
         IHomeAssistantApiManager apiManager,
-        IQueuedObservable<HassEvent> queuedObservable,
         IBackgroundTaskTracker backgroundTaskTracker,
-        IServiceProvider serviceProvider
-    )
+        IServiceProvider serviceProvider,
+        ILogger<AppScopedHaContextProvider> logger)
     {
         _entityStateCache = entityStateCache;
         _hassRunner = hassRunner;
         _apiManager = apiManager;
 
-        // Create ScopedObservables for this app
+        // Create QueuedObservable for this app
         // This makes sure we will unsubscribe when this ContextProvider is Disposed
-        _queuedObservable = queuedObservable;
+        _queuedObservable = new QueuedObservable<HassEvent>(_entityStateCache.AllEvents, logger);
         _backgroundTaskTracker = backgroundTaskTracker;
-
-        _queuedObservable.Initialize(_entityStateCache.AllEvents);
+        _logger = logger;
 
         // The HaRegistry needs a reference to this AppScopedHaContextProvider And we need the reference
         // to the AppScopedHaContextProvider here. Therefore we create it manually providing this
@@ -68,13 +67,15 @@ public IReadOnlyList<Entity> GetAllEntities()
 
     public void CallService(string domain, string service, ServiceTarget? target = null, object? data = null)
     {
-        ObjectDisposedException.ThrowIf(_isDisposed, nameof(AppScopedHaContextProvider));
-        _backgroundTaskTracker.TrackBackgroundTask(_hassRunner.CurrentConnection?.CallServiceAsync(domain, service, data, target.Map(), _tokenSource.Token), "Error in sending event");
+        ObjectDisposedException.ThrowIf(_isDisposed, this);
+        _ = _hassRunner.CurrentConnection ?? throw new InvalidOperationException("No connection to Home Assistant");
+
+        _backgroundTaskTracker.TrackBackgroundTask(_hassRunner.CurrentConnection.CallServiceAsync(domain, service, data, target.Map(), _tokenSource.Token), "Error in sending event");
     }
 
     public async Task<JsonElement?> CallServiceWithResponseAsync(string domain, string service, ServiceTarget? target = null, object? data = null)
     {
-        ObjectDisposedException.ThrowIf(_isDisposed, nameof(AppScopedHaContextProvider));
+        ObjectDisposedException.ThrowIf(_isDisposed, this);
         _ = _hassRunner.CurrentConnection ?? throw new InvalidOperationException("No connection to Home Assistant");
 
         var result = await _hassRunner.CurrentConnection
@@ -94,22 +95,25 @@ public IObservable<StateChange> StateAllChanges()
     public IObservable<Event> Events => _queuedObservable
         .Select(e => e.Map());
 
-
     public void SendEvent(string eventType, object? data = null)
     {
-        ObjectDisposedException.ThrowIf(_isDisposed, nameof(AppScopedHaContextProvider));
+        ObjectDisposedException.ThrowIf(_isDisposed, this);
         _backgroundTaskTracker.TrackBackgroundTask(_apiManager.SendEventAsync(eventType, _tokenSource.Token, data), "Error in sending event");
     }
 
     public async ValueTask DisposeAsync()
     {
-        if (_isDisposed) return;
-        _isDisposed = true;
+        if (_isDisposing) return;
+        _isDisposing = true;
+
+        // The order here is important, we want to allow apps to process their pending events and wait for background tasks to complete
+        // before actually shutting down
+        await _queuedObservable.DisposeAsync().ConfigureAwait(false);
+        await _backgroundTaskTracker.DisposeAsync().ConfigureAwait(false);
 
-        if (!_tokenSource.IsCancellationRequested)
-            await _tokenSource.CancelAsync();
-        //  Wait for all background tasks to complete before disposing the CancellationTokenSource
-        await _backgroundTaskTracker.DisposeAsync();
+        await _tokenSource.CancelAsync().ConfigureAwait(false);
+
+        _isDisposed = true;
         _tokenSource.Dispose();
     }
 }