Proof of concept realm subscriptions via Register

osu.Game/Database/EmptyRealmSet.cs

@@ -0,0 +1,74 @@
+// Copyright (c) ppy Pty Ltd <contact@ppy.sh>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System;
+using System.Collections;
+using System.Collections.Generic;
+using System.Collections.Specialized;
+using System.ComponentModel;
+using Realms;
+using Realms.Schema;
+
+#nullable enable
+
+namespace osu.Game.Database
+{
+    public class EmptyRealmSet<T> : IRealmCollection<T>
+    {
+        private static List<T> emptySet => new List<T>();
+
+        public IEnumerator<T> GetEnumerator()
+        {
+            return emptySet.GetEnumerator();
+        }
+
+        IEnumerator IEnumerable.GetEnumerator()
+        {
+            return ((IEnumerable)emptySet).GetEnumerator();
+        }
+
+        public int Count => emptySet.Count;
+
+        public T this[int index] => emptySet[index];
+
+        public event NotifyCollectionChangedEventHandler? CollectionChanged
+        {
+            add => throw new NotImplementedException();
+            remove => throw new NotImplementedException();
+        }
+
+        public event PropertyChangedEventHandler? PropertyChanged
+        {
+            add => throw new NotImplementedException();
+            remove => throw new NotImplementedException();
+        }
+
+        public int IndexOf(object item)
+        {
+            return emptySet.IndexOf((T)item);
+        }
+
+        public bool Contains(object item)
+        {
+            return emptySet.Contains((T)item);
+        }
+
+        public IRealmCollection<T> Freeze()
+        {
+            throw new NotImplementedException();
+        }
+
+        public IDisposable SubscribeForNotifications(NotificationCallbackDelegate<T> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public bool IsValid => throw new NotImplementedException();
+
+        public Realm Realm => throw new NotImplementedException();
+
+        public ObjectSchema ObjectSchema => throw new NotImplementedException();
+
+        public bool IsFrozen => throw new NotImplementedException();
+    }
+}

osu.Game/Database/RealmContextFactory.cs

@@ -84,7 +84,7 @@ namespace osu.Game.Database
                     Logger.Log(@$"Opened realm ""{context.Config.DatabasePath}"" at version {context.Config.SchemaVersion}");
 
                     // Resubscribe any subscriptions
-                    foreach (var action in subscriptionActions.Keys)
+                    foreach (var action in customSubscriptionActions.Keys)
                         registerSubscription(action);
                 }
 
@@ -233,7 +233,22 @@ namespace osu.Game.Database
             }
         }
 
-        private readonly Dictionary<Func<Realm, IDisposable?>, IDisposable?> subscriptionActions = new Dictionary<Func<Realm, IDisposable?>, IDisposable?>();
+        private readonly Dictionary<Func<Realm, IDisposable?>, IDisposable?> customSubscriptionActions = new Dictionary<Func<Realm, IDisposable?>, IDisposable?>();
+
+        private readonly Dictionary<Func<Realm, IDisposable?>, Action> realmSubscriptionsResetMap = new Dictionary<Func<Realm, IDisposable?>, Action>();
+
+        public IDisposable Register<T>(Func<Realm, IQueryable<T>> query, NotificationCallbackDelegate<T> onChanged)
+            where T : RealmObjectBase
+        {
+            if (!ThreadSafety.IsUpdateThread)
+                throw new InvalidOperationException(@$"{nameof(Register)} must be called from the update thread.");
+
+            realmSubscriptionsResetMap.Add(action, () => onChanged(new EmptyRealmSet<T>(), null, null));
+
+            return Register(action);
+
+            IDisposable? action(Realm realm) => query(realm).QueryAsyncWithNotifications(onChanged);
+        }
 
         /// <summary>
         /// Run work on realm that will be run every time the update thread realm context gets recycled.
@@ -253,10 +268,11 @@ namespace osu.Game.Database
             {
                 lock (contextLock)
                 {
-                    if (subscriptionActions.TryGetValue(action, out var unsubscriptionAction))
+                    if (customSubscriptionActions.TryGetValue(action, out var unsubscriptionAction))
                     {
                         unsubscriptionAction?.Dispose();
-                        subscriptionActions.Remove(action);
+                        customSubscriptionActions.Remove(action);
+                        realmSubscriptionsResetMap.Remove(action);
                     }
                 }
             });
@@ -274,7 +290,7 @@ namespace osu.Game.Database
                 Debug.Assert(!customSubscriptionActions.TryGetValue(action, out var found) || found == null);
 
                 current_thread_subscriptions_allowed.Value = true;
-                subscriptionActions[action] = action(realm);
+                customSubscriptionActions[action] = action(realm);
                 current_thread_subscriptions_allowed.Value = false;
             }
         }
@@ -285,10 +301,13 @@ namespace osu.Game.Database
         /// </summary>
         private void unregisterAllSubscriptions()
         {
-            foreach (var action in subscriptionActions)
+            foreach (var action in realmSubscriptionsResetMap.Values)
+                action();
+
+            foreach (var action in customSubscriptionActions)
             {
                 action.Value?.Dispose();
-                subscriptionActions[action.Key] = null;
+                customSubscriptionActions[action.Key] = null;
             }
         }
 

osu.Game/Screens/Select/BeatmapCarousel.cs

@@ -190,7 +190,7 @@ namespace osu.Game.Screens.Select
         {
             base.LoadComplete();
 
-            subscriptionSets = realmFactory.Register(realm => getBeatmapSets(realm).QueryAsyncWithNotifications(beatmapSetsChanged));
+            subscriptionSets = realmFactory.Register(getBeatmapSets, beatmapSetsChanged);
             subscriptionBeatmaps = realmFactory.Register(realm => realm.All<BeatmapInfo>().Where(b => !b.Hidden).QueryAsyncWithNotifications(beatmapsChanged));
 
             // Can't use main subscriptions because we can't lookup deleted indices.
@@ -274,7 +274,7 @@ namespace osu.Game.Screens.Select
             }
         }
 
-        private IRealmCollection<BeatmapSetInfo> getBeatmapSets(Realm realm) => realm.All<BeatmapSetInfo>().Where(s => !s.DeletePending && !s.Protected).AsRealmCollection();
+        private IQueryable<BeatmapSetInfo> getBeatmapSets(Realm realm) => realm.All<BeatmapSetInfo>().Where(s => !s.DeletePending && !s.Protected);
 
         public void RemoveBeatmapSet(BeatmapSetInfo beatmapSet) =>
             removeBeatmapSet(beatmapSet.ID);

osu.Game/Screens/Spectate/SpectatorScreen.cs

@@ -17,6 +17,7 @@ using osu.Game.Online.Spectator;
 using osu.Game.Replays;
 using osu.Game.Rulesets;
 using osu.Game.Scoring;
+using Realms;
 
 namespace osu.Game.Screens.Spectate
 {
@@ -79,23 +80,21 @@ namespace osu.Game.Screens.Spectate
                 playingUserStates.BindTo(spectatorClient.PlayingUserStates);
                 playingUserStates.BindCollectionChanged(onPlayingUserStatesChanged, true);
 
-                realmSubscription = realmContextFactory.Register(realm =>
+                realmSubscription = realmContextFactory.Register(
-                    realm.All<BeatmapSetInfo>()
+                    realm => realm.All<BeatmapSetInfo>().Where(s => !s.DeletePending), beatmapsChanged);
-                         .Where(s => !s.DeletePending)
-                         .QueryAsyncWithNotifications((items, changes, ___) =>
-                         {
-                             if (changes?.InsertedIndices == null)
-                                 return;
-
-                             foreach (int c in changes.InsertedIndices)
-                                 beatmapUpdated(items[c]);
-                         }));
 
                 foreach ((int id, var _) in userMap)
                     spectatorClient.WatchUser(id);
             }));
         }
 
+        private void beatmapsChanged(IRealmCollection<BeatmapSetInfo> items, ChangeSet changes, Exception ___)
+        {
+            if (changes?.InsertedIndices == null) return;
+
+            foreach (int c in changes.InsertedIndices) beatmapUpdated(items[c]);
+        }
+
         private void beatmapUpdated(BeatmapSetInfo beatmapSet)
         {
             foreach ((int userId, _) in userMap)