Change polling flow to not require initial separate call

Author: peppy

osu.Server.QueueProcessor/BeatmapStatusWatcher.cs

@@ -19,40 +19,40 @@ public static class BeatmapStatusWatcher
         /// Start a background task which will poll for beatmap sets with updates.
         /// </summary>
         /// <remarks>
-        /// Prior to polling, a blocking call to <see cref="GetUpdatedBeatmapSets"/> is required to ensure no initial updates are missed.
         /// The general flow of usage should be:
         ///
-        /// // before doing anything else
-        /// var updates = await GetUpdatedBeatmapSets();
-        /// // can now query and cache beatmaps.
-        /// StartPolling(updates, callback);
+        /// // before doing anything else, start polling.
+        /// // it's important to await the completion of this operation to ensure no updates are missed.
+        /// using var pollingOperation = await StartPollingAsync(updates, callback);
         ///
         /// void callback(BeatmapUpdates u)
         /// {
-        ///     foreach (int id in u.BeatmapSetIDs)
+        ///     foreach (int beatmapSetId in u.BeatmapSetIDs)
         ///     {
-        ///         // invalidate `id`
+        ///         // invalidate anything related to `beatmapSetId`
         ///     }
         /// }
         /// </remarks>
-        /// <param name="initialUpdates">The response from an initial call to <see cref="GetUpdatedBeatmapSets"/>.</param>
         /// <param name="callback">A callback to receive information about any updated beatmap sets.</param>
         /// <param name="pollMilliseconds">The number of milliseconds to wait between polls. Starts counting from response of previous poll.</param>
         /// <param name="limit">The maximum number of beatmap sets to return in a single response.</param>
         /// <returns>An <see cref="IDisposable"/> that should be disposed of to stop polling.</returns>
-        public static IDisposable StartPolling(BeatmapUpdates initialUpdates, Action<BeatmapUpdates> callback, int pollMilliseconds = 10000, int limit = 50) =>
-            new PollingBeatmapStatusWatcher(callback, pollMilliseconds, limit);
+        public static async Task<IDisposable> StartPollingAsync(Action<BeatmapUpdates> callback, int pollMilliseconds = 10000, int limit = 50)
+        {
+            var initialUpdates = await GetUpdatedBeatmapSetsAsync(limit: limit);
+            return new PollingBeatmapStatusWatcher(initialUpdates.LastProcessedQueueID, callback, pollMilliseconds, limit);
+        }
 
         /// <summary>
         /// Check for any beatmap sets with updates since the provided queue ID.
-        /// Should be called on a regular basis. See <see cref="StartPolling"/> for automatic polling after the first call.
+        /// Should be called on a regular basis. See <see cref="StartPollingAsync"/> for automatic polling after the first call.
         /// </summary>
         /// <param name="lastQueueId">The last checked queue ID, ie <see cref="BeatmapUpdates.LastProcessedQueueID"/>.</param>
         /// <param name="limit">The maximum number of beatmap sets to return in a single response.</param>
         /// <returns>A response containing information about any updated beatmap sets.</returns>
-        public static async Task<BeatmapUpdates> GetUpdatedBeatmapSets(int? lastQueueId, int limit = 50)
+        public static async Task<BeatmapUpdates> GetUpdatedBeatmapSetsAsync(int? lastQueueId = null, int limit = 50)
         {
-            MySqlConnection connection = await DatabaseAccess.GetConnectionAsync();
+            using MySqlConnection connection = await DatabaseAccess.GetConnectionAsync();
 
             if (lastQueueId.HasValue)
             {
@@ -86,57 +86,52 @@ public class bss_process_queue_item
             public int beatmapset_id;
         }
 
-        public record BeatmapUpdates
+        private class PollingBeatmapStatusWatcher : IDisposable
         {
-            public required int[] BeatmapSetIDs { get; init; }
-            public required int LastProcessedQueueID { get; init; }
-        }
-    }
-
-    public class PollingBeatmapStatusWatcher : IDisposable
-    {
-        private readonly Action<BeatmapStatusWatcher.BeatmapUpdates> callback;
+            private readonly Action<BeatmapUpdates> callback;
 
-        private readonly int pollMilliseconds;
-        private readonly int limit;
+            private readonly int pollMilliseconds;
+            private readonly int limit;
 
-        private int? lastQueueId;
-        private readonly CancellationTokenSource cts;
+            private int lastQueueId;
+            private readonly CancellationTokenSource cts;
 
-        public PollingBeatmapStatusWatcher(Action<BeatmapStatusWatcher.BeatmapUpdates> callback, int pollMilliseconds, int limit = 50)
-        {
-            this.pollMilliseconds = pollMilliseconds;
-            this.limit = limit;
-            this.callback = callback;
+            public PollingBeatmapStatusWatcher(int initialQueueId, Action<BeatmapUpdates> callback, int pollMilliseconds, int limit = 50)
+            {
+                this.lastQueueId = initialQueueId;
+                this.pollMilliseconds = pollMilliseconds;
+                this.limit = limit;
+                this.callback = callback;
 
-            cts = new CancellationTokenSource();
+                cts = new CancellationTokenSource();
 
-            _ = poll();
-        }
+                _ = poll();
+            }
 
-        private async Task poll()
-        {
-            try
+            private async Task poll()
             {
-                var result = await BeatmapStatusWatcher.GetUpdatedBeatmapSets(lastQueueId, limit);
+                try
+                {
+                    var result = await BeatmapStatusWatcher.GetUpdatedBeatmapSetsAsync(lastQueueId, limit);
 
-                lastQueueId = result.LastProcessedQueueID;
-                if (result.BeatmapSetIDs.Length > 0)
-                    callback(result);
+                    lastQueueId = result.LastProcessedQueueID;
+                    if (result.BeatmapSetIDs.Length > 0)
+                        callback(result);
+                }
+                catch (Exception e)
+                {
+                    Console.WriteLine($"Poll failed with {e}.");
+                    await Task.Delay(1000);
+                }
+
+                _ = Task.Delay(pollMilliseconds, cts.Token).ContinueWith(_ => poll(), cts.Token);
             }
-            catch (Exception e)
+
+            public void Dispose()
             {
-                Console.WriteLine($"Poll failed with {e}.");
-                await Task.Delay(1000);
+                cts.Cancel();
+                cts.Dispose();
             }
-
-            _ = Task.Delay(pollMilliseconds, cts.Token).ContinueWith(_ => poll(), cts.Token);
-        }
-
-        public void Dispose()
-        {
-            cts.Cancel();
-            cts.Dispose();
         }
     }
 }

osu.Server.QueueProcessor/BeatmapUpdates.cs

@@ -0,0 +1,11 @@
+// Copyright (c) ppy Pty Ltd <[email protected]>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+namespace osu.Server.QueueProcessor
+{
+    public record BeatmapUpdates
+    {
+        public required int[] BeatmapSetIDs { get; init; }
+        public required int LastProcessedQueueID { get; init; }
+    }
+}