Fixed event/observer issues

MoonlightServers.Daemon/MoonlightServers.Daemon.csproj

@@ -9,7 +9,7 @@
     <ItemGroup>
         <PackageReference Include="Docker.DotNet" Version="3.125.15" />
         <PackageReference Include="Microsoft.AspNetCore.SignalR" Version="1.2.0" />
-        <PackageReference Include="MoonCore" Version="1.9.3" />
+        <PackageReference Include="MoonCore" Version="1.9.4" />
         <PackageReference Include="MoonCore.Extended" Version="1.3.6" />
         <PackageReference Include="MoonCore.Unix" Version="1.0.8" />
         <PackageReference Include="SharpZipLib" Version="1.4.2" />

MoonlightServers.Daemon/ServerSys/Abstractions/Server.cs

@@ -1,6 +1,6 @@
-using System.Reactive.Linq;
+using MoonCore.Observability;
-using System.Reactive.Subjects;
 using MoonlightServers.Daemon.Extensions;
+using MoonlightServers.Daemon.Helpers;
 using MoonlightServers.Daemon.ServerSystem;
 using Stateless;
 
@@ -8,17 +8,17 @@ namespace MoonlightServers.Daemon.ServerSys.Abstractions;
 
 public class Server : IAsyncDisposable
 {
-    public IConsole Console { get; private set; }
+    public IConsole Console { get; }
-    public IFileSystem FileSystem { get; private set; }
+    public IFileSystem FileSystem { get; }
-    public IInstaller Installer { get; private set; }
+    public IInstaller Installer { get; }
-    public IProvisioner Provisioner { get; private set; }
+    public IProvisioner Provisioner { get; }
-    public IRestorer Restorer { get; private set; }
+    public IRestorer Restorer { get; }
-    public IStatistics Statistics { get; private set; }
+    public IStatistics Statistics { get; }
     public StateMachine<ServerState, ServerTrigger> StateMachine { get; private set; }
-    public ServerContext Context { get; private set; }
+    public ServerContext Context { get; }
-    public IAsyncObservable<ServerState> OnState => OnStateSubject.ToAsyncObservable();
+    public IAsyncObservable<ServerState> OnState => OnStateSubject;
 
-    private readonly Subject<ServerState> OnStateSubject = new();
+    private readonly EventSubject<ServerState> OnStateSubject = new();
     private readonly ILogger<Server> Logger;
 
     private IAsyncDisposable? ProvisionExitSubscription;
@@ -80,22 +80,22 @@ public class Server : IAsyncDisposable
         CreateStateMachine(restoredState);
 
         // Setup event handling
-        ProvisionExitSubscription = await Provisioner.OnExited.SubscribeAsync(async o =>
+        ProvisionExitSubscription = await Provisioner.OnExited.SubscribeEventAsync(async _ =>
-        {
+            await StateMachine.FireAsync(ServerTrigger.Exited)
-            await StateMachine.FireAsync(ServerTrigger.Exited);
+        );
-        });
 
-        InstallerExitSubscription = await Installer.OnExited.SubscribeAsync(async o =>
+        InstallerExitSubscription = await Installer.OnExited.SubscribeEventAsync(async _ =>
-        {
+            await StateMachine.FireAsync(ServerTrigger.Exited)
-            await StateMachine.FireAsync(ServerTrigger.Exited);
+        );
-        });
     }
 
     private void CreateStateMachine(ServerState initialState)
     {
         StateMachine = new StateMachine<ServerState, ServerTrigger>(initialState, FiringMode.Queued);
 
-        StateMachine.OnTransitioned(transition => OnStateSubject.OnNext(transition.Destination));
+        StateMachine.OnTransitionedAsync(async transition
+            => await OnStateSubject.OnNextAsync(transition.Destination)
+        );
 
         // Configure basic state machine flow
 
@@ -187,6 +187,9 @@ public class Server : IAsyncDisposable
         catch (Exception e)
         {
             Logger.LogError(e, "An error occured while starting the server");
+            await Console.WriteToMoonlight("Failed to start the server. More details can be found in the daemon logs");
+            
+            await StateMachine.FireAsync(ServerTrigger.FailSafe);
         }
     }
 
@@ -204,6 +207,8 @@ public class Server : IAsyncDisposable
     }
 
     private async Task HandleInstall()
+    {
+        try
         {
             Logger.LogDebug("Installing");
 
@@ -226,6 +231,14 @@ public class Server : IAsyncDisposable
 
             await Installer.Start();
         }
+        catch (Exception e)
+        {
+            Logger.LogError(e, "An error occured while starting installation");
+            await Console.WriteToMoonlight("An error occured while starting installation");
+
+            await StateMachine.FireAsync(ServerTrigger.FailSafe);
+        }
+    }
 
     private async Task HandleInstallExit()
     {

MoonlightServers.Daemon/ServerSys/Implementations/DockerConsole.cs

@@ -4,17 +4,19 @@ using System.Text;
 using Docker.DotNet;
 using Docker.DotNet.Models;
 using MoonCore.Helpers;
+using MoonCore.Observability;
+using MoonlightServers.Daemon.Helpers;
 using MoonlightServers.Daemon.ServerSys.Abstractions;
 
 namespace MoonlightServers.Daemon.ServerSys.Implementations;
 
 public class DockerConsole : IConsole
 {
-    public IAsyncObservable<string> OnOutput => OnOutputSubject.ToAsyncObservable();
+    public IAsyncObservable<string> OnOutput => OnOutputSubject;
-    public IAsyncObservable<string> OnInput => OnInputSubject.ToAsyncObservable();
+    public IAsyncObservable<string> OnInput => OnInputSubject;
 
-    private readonly Subject<string> OnOutputSubject = new();
+    private readonly EventSubject<string> OnOutputSubject = new();
-    private readonly Subject<string> OnInputSubject = new();
+    private readonly EventSubject<string> OnInputSubject = new();
 
     private readonly ConcurrentList<string> OutputCache = new();
     private readonly DockerClient DockerClient;
@@ -140,15 +142,14 @@ public class DockerConsole : IConsole
         return Task.CompletedTask;
     }
 
-    public Task WriteToOutput(string content)
+    public async Task WriteToOutput(string content)
     {
         OutputCache.Add(content);
 
         if (OutputCache.Count > 250) // TODO: Config
             OutputCache.RemoveRange(0, 100);
 
-        OnOutputSubject.OnNext(content);
+        await OnOutputSubject.OnNextAsync(content);
-        return Task.CompletedTask;
     }
 
     public async Task WriteToInput(string content)
@@ -164,6 +165,8 @@ public class DockerConsole : IConsole
             contentBuffer.Length,
             Cts.Token
         );
+        
+        await OnInputSubject.OnNextAsync(content);
     }
 
     public async Task WriteToMoonlight(string content)

MoonlightServers.Daemon/ServerSys/Implementations/DockerInstaller.cs

@@ -2,7 +2,10 @@ using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using Docker.DotNet;
 using Docker.DotNet.Models;
+using MoonCore.Observability;
 using MoonlightServers.Daemon.Configuration;
+using MoonlightServers.Daemon.Extensions;
+using MoonlightServers.Daemon.Helpers;
 using MoonlightServers.Daemon.Mappers;
 using MoonlightServers.Daemon.ServerSys.Abstractions;
 using MoonlightServers.Daemon.Services;
@@ -11,10 +14,10 @@ namespace MoonlightServers.Daemon.ServerSys.Implementations;
 
 public class DockerInstaller : IInstaller
 {
-    public IAsyncObservable<object> OnExited => OnExitedSubject.ToAsyncObservable();
+    public IAsyncObservable<object> OnExited => OnExitedSubject;
     public bool IsRunning { get; private set; } = false;
 
-    private readonly Subject<Message> OnExitedSubject = new();
+    private readonly EventSubject<Message> OnExitedSubject = new();
 
     private readonly ILogger<DockerInstaller> Logger;
     private readonly DockerEventService EventService;
@@ -64,7 +67,7 @@ public class DockerInstaller : IInstaller
         
         ContainerEventSubscription = await EventService
             .OnContainerEvent
-            .SubscribeAsync(HandleContainerEvent);
+            .SubscribeEventAsync(HandleContainerEvent);
         
         // Check for any already existing runtime container to reclaim
         Logger.LogDebug("Searching for orphan container to reclaim");
@@ -82,19 +85,17 @@ public class DockerInstaller : IInstaller
         }
     }
     
-    private ValueTask HandleContainerEvent(Message message)
+    private async ValueTask HandleContainerEvent(Message message)
     {
         // Only handle events for our own container
         if (message.ID != ContainerId)
-            return ValueTask.CompletedTask;
+            return;
 
         // Only handle die events
         if (message.Action != "die")
-            return ValueTask.CompletedTask;
+            return;
 
-        OnExitedSubject.OnNext(message);
+        await OnExitedSubject.OnNextAsync(message);
-
-        return ValueTask.CompletedTask;
     }
 
     public Task Sync()

MoonlightServers.Daemon/ServerSys/Implementations/DockerProvisioner.cs

@@ -3,6 +3,9 @@ using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using Docker.DotNet;
 using Docker.DotNet.Models;
+using MoonCore.Observability;
+using MoonlightServers.Daemon.Extensions;
+using MoonlightServers.Daemon.Helpers;
 using MoonlightServers.Daemon.Mappers;
 using MoonlightServers.Daemon.ServerSys.Abstractions;
 using MoonlightServers.Daemon.Services;
@@ -11,7 +14,7 @@ namespace MoonlightServers.Daemon.ServerSys.Implementations;
 
 public class DockerProvisioner : IProvisioner
 {
-    public IAsyncObservable<object> OnExited => OnExitedSubject.ToAsyncObservable();
+    public IAsyncObservable<object> OnExited => OnExitedSubject;
     public bool IsProvisioned { get; private set; }
 
     private readonly DockerClient DockerClient;
@@ -23,7 +26,7 @@ public class DockerProvisioner : IProvisioner
     private readonly ServerConfigurationMapper Mapper;
     private readonly IFileSystem FileSystem;
 
-    private Subject<object> OnExitedSubject = new();
+    private EventSubject<object> OnExitedSubject = new();
 
     private string? ContainerId;
     private string ContainerName;
@@ -56,7 +59,7 @@ public class DockerProvisioner : IProvisioner
 
         ContainerEventSubscription = await EventService
             .OnContainerEvent
-            .SubscribeAsync(HandleContainerEvent);
+            .SubscribeEventAsync(HandleContainerEvent);
         
         // Check for any already existing runtime container to reclaim
         Logger.LogDebug("Searching for orphan container to reclaim");
@@ -74,19 +77,17 @@ public class DockerProvisioner : IProvisioner
         }
     }
 
-    private ValueTask HandleContainerEvent(Message message)
+    private async ValueTask HandleContainerEvent(Message message)
     {
         // Only handle events for our own container
         if (message.ID != ContainerId)
-            return ValueTask.CompletedTask;
+            return;
 
         // Only handle die events
         if (message.Action != "die")
-            return ValueTask.CompletedTask;
+            return;
 
-        OnExitedSubject.OnNext(message);
+        await OnExitedSubject.OnNextAsync(message);
-
-        return ValueTask.CompletedTask;
     }
 
     public Task Sync()

MoonlightServers.Daemon/ServerSys/Implementations/DockerStatistics.cs

@@ -1,14 +1,14 @@
-using System.Reactive.Linq;
+using MoonCore.Observability;
-using System.Reactive.Subjects;
+using MoonlightServers.Daemon.Helpers;
 using MoonlightServers.Daemon.ServerSys.Abstractions;
 
 namespace MoonlightServers.Daemon.ServerSys.Implementations;
 
 public class DockerStatistics : IStatistics
 {
-    public IAsyncObservable<ServerStats> OnStats => OnStatsSubject.ToAsyncObservable();
+    public IAsyncObservable<ServerStats> OnStats => OnStatsSubject;
 
-    private readonly Subject<ServerStats> OnStatsSubject = new();
+    private readonly EventSubject<ServerStats> OnStatsSubject = new();
     
     public Task Initialize()
         => Task.CompletedTask;

MoonlightServers.Daemon/Services/DockerEventService.cs

@@ -3,6 +3,8 @@ using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using Docker.DotNet;
 using Docker.DotNet.Models;
+using MoonCore.Observability;
+using MoonlightServers.Daemon.Helpers;
 
 namespace MoonlightServers.Daemon.Services;
 
@@ -11,13 +13,13 @@ public class DockerEventService : BackgroundService
     private readonly ILogger<DockerEventService> Logger;
     private readonly DockerClient DockerClient;
 
-    public IAsyncObservable<Message> OnContainerEvent => OnContainerSubject.ToAsyncObservable().ObserveOn(TaskPoolAsyncScheduler.Default);
+    public IAsyncObservable<Message> OnContainerEvent => OnContainerSubject;
-    public IAsyncObservable<Message> OnImageEvent => OnImageSubject.ToAsyncObservable().ObserveOn(TaskPoolAsyncScheduler.Default);
+    public IAsyncObservable<Message> OnImageEvent => OnImageSubject;
-    public IAsyncObservable<Message> OnNetworkEvent =>  OnNetworkSubject.ToAsyncObservable().ObserveOn(TaskPoolAsyncScheduler.Default);
+    public IAsyncObservable<Message> OnNetworkEvent =>  OnNetworkSubject;
     
-    private readonly Subject<Message> OnContainerSubject = new();
+    private readonly EventSubject<Message> OnContainerSubject = new();
-    private readonly Subject<Message> OnImageSubject = new();
+    private readonly EventSubject<Message> OnImageSubject = new();
-    private readonly Subject<Message> OnNetworkSubject = new();
+    private readonly EventSubject<Message> OnNetworkSubject = new();
 
     public DockerEventService(
         ILogger<DockerEventService> logger,
@@ -38,22 +40,22 @@ public class DockerEventService : BackgroundService
             {
                 await DockerClient.System.MonitorEventsAsync(
                     new ContainerEventsParameters(),
-                    new Progress<Message>(message =>
+                    new Progress<Message>(async message =>
                     {
                         try
                         {
                             switch (message.Type)
                             {
                                 case "container":
-                                    OnContainerSubject.OnNext(message);
+                                    await OnContainerSubject.OnNextAsync(message);
                                     break;
 
                                 case "image":
-                                    OnImageSubject.OnNext(message);
+                                    await OnImageSubject.OnNextAsync(message);
                                     break;
 
                                 case "network":
-                                    OnNetworkSubject.OnNext(message);
+                                    await OnNetworkSubject.OnNextAsync(message);
                                     break;
                             }
                         }

MoonlightServers.Daemon/Startup.cs

@@ -11,7 +11,9 @@ using MoonCore.Extended.Extensions;
 using MoonCore.Extensions;
 using MoonCore.Helpers;
 using MoonCore.Logging;
+using MoonCore.Observability;
 using MoonlightServers.Daemon.Configuration;
+using MoonlightServers.Daemon.Extensions;
 using MoonlightServers.Daemon.Helpers;
 using MoonlightServers.Daemon.Http.Hubs;
 using MoonlightServers.Daemon.Mappers;
@@ -108,9 +110,13 @@ public class Startup
                 var server = factory.CreateServer(config);
 
                 await using var consoleSub = await server.Console.OnOutput
-                    .SubscribeAsync(Console.Write);
+                    .SubscribeEventAsync(line =>
+                    {
+                        Console.Write(line);
+                        return ValueTask.CompletedTask;
+                    });
 
-                await using var stateSub = await server.OnState.SubscribeAsync(state =>
+                await using var stateSub = await server.OnState.SubscribeEventAsync(state =>
                 {
                     Console.WriteLine($"State: {state}");
                     return ValueTask.CompletedTask;