refactor: use Channels as queueing mechanism for periodic websocket messages (#11092)

3 files changed, 39 insertions, 31 deletions

diff --git a/Jellyfin.Api/WebSocketListeners/ActivityLogWebSocketListener.cs b/Jellyfin.Api/WebSocketListeners/ActivityLogWebSocketListener.cs
index ba228cb00..99516e938 100644
--- a/Jellyfin.Api/WebSocketListeners/ActivityLogWebSocketListener.cs
+++ b/Jellyfin.Api/WebSocketListeners/ActivityLogWebSocketListener.cs
@@ -20,6 +20,8 @@ public class ActivityLogWebSocketListener : BasePeriodicWebSocketListener<Activi
     /// </summary>
     private readonly IActivityManager _activityManager;
 
+    private bool _disposed;
+
     /// <summary>
     /// Initializes a new instance of the <see cref="ActivityLogWebSocketListener"/> class.
     /// </summary>
@@ -51,14 +53,15 @@ public class ActivityLogWebSocketListener : BasePeriodicWebSocketListener<Activi
     }
 
     /// <inheritdoc />
-    protected override void Dispose(bool dispose)
+    protected override async ValueTask DisposeAsyncCore()
     {
-        if (dispose)
+        if (!_disposed)
         {
             _activityManager.EntryCreated -= OnEntryCreated;
+            _disposed = true;
         }
 
-        base.Dispose(dispose);
+        await base.DisposeAsyncCore().ConfigureAwait(false);
     }
 
     /// <summary>
@@ -75,8 +78,8 @@ public class ActivityLogWebSocketListener : BasePeriodicWebSocketListener<Activi
         base.Start(message);
     }
 
-    private async void OnEntryCreated(object? sender, GenericEventArgs<ActivityLogEntry> e)
+    private void OnEntryCreated(object? sender, GenericEventArgs<ActivityLogEntry> e)
     {
-        await SendData(true).ConfigureAwait(false);
+        SendData(true);
     }
 }
diff --git a/Jellyfin.Api/WebSocketListeners/ScheduledTasksWebSocketListener.cs b/Jellyfin.Api/WebSocketListeners/ScheduledTasksWebSocketListener.cs
index 37c108d5a..dd9286210 100644
--- a/Jellyfin.Api/WebSocketListeners/ScheduledTasksWebSocketListener.cs
+++ b/Jellyfin.Api/WebSocketListeners/ScheduledTasksWebSocketListener.cs
@@ -20,6 +20,8 @@ public class ScheduledTasksWebSocketListener : BasePeriodicWebSocketListener<IEn
     /// <value>The task manager.</value>
     private readonly ITaskManager _taskManager;
 
+    private bool _disposed;
+
     /// <summary>
     /// Initializes a new instance of the <see cref="ScheduledTasksWebSocketListener"/> class.
     /// </summary>
@@ -56,31 +58,32 @@ public class ScheduledTasksWebSocketListener : BasePeriodicWebSocketListener<IEn
     }
 
     /// <inheritdoc />
-    protected override void Dispose(bool dispose)
+    protected override async ValueTask DisposeAsyncCore()
     {
-        if (dispose)
+        if (!_disposed)
         {
             _taskManager.TaskExecuting -= OnTaskExecuting;
             _taskManager.TaskCompleted -= OnTaskCompleted;
+            _disposed = true;
         }
 
-        base.Dispose(dispose);
+        await base.DisposeAsyncCore().ConfigureAwait(false);
     }
 
-    private async void OnTaskCompleted(object? sender, TaskCompletionEventArgs e)
+    private void OnTaskCompleted(object? sender, TaskCompletionEventArgs e)
     {
         e.Task.TaskProgress -= OnTaskProgress;
-        await SendData(true).ConfigureAwait(false);
+        SendData(true);
     }
 
-    private async void OnTaskExecuting(object? sender, GenericEventArgs<IScheduledTaskWorker> e)
+    private void OnTaskExecuting(object? sender, GenericEventArgs<IScheduledTaskWorker> e)
     {
-        await SendData(true).ConfigureAwait(false);
+        SendData(true);
         e.Argument.TaskProgress += OnTaskProgress;
     }
 
-    private async void OnTaskProgress(object? sender, GenericEventArgs<double> e)
+    private void OnTaskProgress(object? sender, GenericEventArgs<double> e)
     {
-        await SendData(false).ConfigureAwait(false);
+        SendData(false);
     }
 }
diff --git a/Jellyfin.Api/WebSocketListeners/SessionInfoWebSocketListener.cs b/Jellyfin.Api/WebSocketListeners/SessionInfoWebSocketListener.cs
index 3c2b86142..a6cfe4d56 100644
--- a/Jellyfin.Api/WebSocketListeners/SessionInfoWebSocketListener.cs
+++ b/Jellyfin.Api/WebSocketListeners/SessionInfoWebSocketListener.cs
@@ -16,6 +16,7 @@ namespace Jellyfin.Api.WebSocketListeners;
 public class SessionInfoWebSocketListener : BasePeriodicWebSocketListener<IEnumerable<SessionInfo>, WebSocketListenerState>
 {
     private readonly ISessionManager _sessionManager;
+    private bool _disposed;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="SessionInfoWebSocketListener"/> class.
@@ -55,9 +56,9 @@ public class SessionInfoWebSocketListener : BasePeriodicWebSocketListener<IEnume
     }
 
     /// <inheritdoc />
-    protected override void Dispose(bool dispose)
+    protected override async ValueTask DisposeAsyncCore()
     {
-        if (dispose)
+        if (!_disposed)
         {
             _sessionManager.SessionStarted -= OnSessionManagerSessionStarted;
             _sessionManager.SessionEnded -= OnSessionManagerSessionEnded;
@@ -66,9 +67,10 @@ public class SessionInfoWebSocketListener : BasePeriodicWebSocketListener<IEnume
             _sessionManager.PlaybackProgress -= OnSessionManagerPlaybackProgress;
             _sessionManager.CapabilitiesChanged -= OnSessionManagerCapabilitiesChanged;
             _sessionManager.SessionActivity -= OnSessionManagerSessionActivity;
+            _disposed = true;
         }
 
-        base.Dispose(dispose);
+        await base.DisposeAsyncCore().ConfigureAwait(false);
     }
 
     /// <summary>
@@ -85,38 +87,38 @@ public class SessionInfoWebSocketListener : BasePeriodicWebSocketListener<IEnume
         base.Start(message);
     }
 
-    private async void OnSessionManagerSessionActivity(object? sender, SessionEventArgs e)
+    private void OnSessionManagerSessionActivity(object? sender, SessionEventArgs e)
     {
-        await SendData(false).ConfigureAwait(false);
+        SendData(false);
     }
 
-    private async void OnSessionManagerCapabilitiesChanged(object? sender, SessionEventArgs e)
+    private void OnSessionManagerCapabilitiesChanged(object? sender, SessionEventArgs e)
     {
-        await SendData(true).ConfigureAwait(false);
+        SendData(true);
     }
 
-    private async void OnSessionManagerPlaybackProgress(object? sender, PlaybackProgressEventArgs e)
+    private void OnSessionManagerPlaybackProgress(object? sender, PlaybackProgressEventArgs e)
     {
-        await SendData(!e.IsAutomated).ConfigureAwait(false);
+        SendData(!e.IsAutomated);
     }
 
-    private async void OnSessionManagerPlaybackStopped(object? sender, PlaybackStopEventArgs e)
+    private void OnSessionManagerPlaybackStopped(object? sender, PlaybackStopEventArgs e)
     {
-        await SendData(true).ConfigureAwait(false);
+        SendData(true);
     }
 
-    private async void OnSessionManagerPlaybackStart(object? sender, PlaybackProgressEventArgs e)
+    private void OnSessionManagerPlaybackStart(object? sender, PlaybackProgressEventArgs e)
     {
-        await SendData(true).ConfigureAwait(false);
+        SendData(true);
     }
 
-    private async void OnSessionManagerSessionEnded(object? sender, SessionEventArgs e)
+    private void OnSessionManagerSessionEnded(object? sender, SessionEventArgs e)
     {
-        await SendData(true).ConfigureAwait(false);
+        SendData(true);
     }
 
-    private async void OnSessionManagerSessionStarted(object? sender, SessionEventArgs e)
+    private void OnSessionManagerSessionStarted(object? sender, SessionEventArgs e)
     {
-        await SendData(true).ConfigureAwait(false);
+        SendData(true);
     }
 }