Merge pull request #94 from Nexus-Mods/behavior-subject-fixes

Behavior subject fixes

Author: halgari

CHANGELOG.md

@@ -1,6 +1,10 @@
 ## Changelog
 
-### 0.9.81 - 19/09/2024
+### 0.9.82 - 12/09/2024
+* Fix a O(n) issue caused by Rx storing observers in a ImmutableList inside a `BehaviorSubject`. Switched to using R3 internally. Over 
+time Rx's uses will be replaced with R3 to avoid these and several other issues
+
+### 0.9.81 - 9/09/2024
 * Fix a bug the source generators when trying to use HashedBlobAttributes
 
 ### 0.9.80 - 22/08/2024

src/NexusMods.MnemonicDB.Abstractions/IDatomStore.cs

@@ -1,10 +1,10 @@
 using System;
 using System.Collections.Generic;
-using System.Threading;
 using System.Threading.Tasks;
 using NexusMods.MnemonicDB.Abstractions.IndexSegments;
 using NexusMods.MnemonicDB.Abstractions.Internals;
 using NexusMods.MnemonicDB.Abstractions.TxFunctions;
+using R3;
 
 namespace NexusMods.MnemonicDB.Abstractions;
 
@@ -17,7 +17,7 @@ public interface IDatomStore : IDisposable
     ///     An observable of the transaction log, for getting the latest changes to the store. This observable
     /// will always start with the most recent value, so there is no reason to use `StartWith` or `Replay` on it.
     /// </summary>
-    public IObservable<IDb> TxLog { get; }
+    public Observable<IDb> TxLog { get; }
 
     /// <summary>
     ///     Gets the latest transaction id found in the log.

src/NexusMods.MnemonicDB.Abstractions/NexusMods.MnemonicDB.Abstractions.csproj

@@ -8,6 +8,7 @@
         <PackageReference Include="DynamicData" Version="8.4.1"/>
         <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.1"/>
         <PackageReference Include="NexusMods.Paths" Version="0.9.5" />
+        <PackageReference Include="R3" Version="1.2.8" />
         <PackageReference Include="System.IO.Hashing" Version="8.0.0" />
         <PackageReference Include="TransparentValueObjects" Version="1.0.1" PrivateAssets="all" OutputItemType="Analyzer" ReferenceOutputAssembly="false"/>
         <PackageReference Update="JetBrains.Annotations" Version="2023.3.0"/>

src/NexusMods.MnemonicDB/Connection.cs

@@ -1,7 +1,6 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
-using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using System.Threading;
 using System.Threading.Tasks;
@@ -13,6 +12,9 @@
 using NexusMods.MnemonicDB.Abstractions.Query;
 using NexusMods.MnemonicDB.Abstractions.TxFunctions;
 using NexusMods.MnemonicDB.Storage;
+using R3;
+using Observable = System.Reactive.Linq.Observable;
+using ObservableExtensions = R3.ObservableExtensions;
 
 namespace NexusMods.MnemonicDB;
 
@@ -25,7 +27,7 @@ public class Connection : IConnection
     private readonly Dictionary<Symbol, IAttribute> _declaredAttributes;
     private readonly ILogger<Connection> _logger;
 
-    private BehaviorSubject<IDb> _dbStream;
+    private R3.BehaviorSubject<IDb> _dbStream;
     private IDisposable? _dbStreamDisposable;
     private readonly IAnalyzer[] _analyzers;
 
@@ -38,19 +40,19 @@ public Connection(ILogger<Connection> logger, IDatomStore store, IServiceProvide
         _logger = logger;
         _declaredAttributes = declaredAttributes.ToDictionary(a => a.Id);
         _store = store;
-        _dbStream = new BehaviorSubject<IDb>(default!);
+        _dbStream = new R3.BehaviorSubject<IDb>(default!);
         _analyzers = analyzers.ToArray();
         Bootstrap();
     }
 
     /// <summary>
     /// Scrubs the transaction stream so that we only ever move forward and never repeat transactions
     /// </summary>
-    private IObservable<Db> ProcessUpdate(IObservable<IDb> dbStream)
+    private R3.Observable<Db> ProcessUpdate(R3.Observable<IDb> dbStream)
     {
         IDb? prev = null;
 
-        return Observable.Create((IObserver<Db> observer) =>
+        return R3.Observable.Create((Observer<Db> observer) =>
         {
             return dbStream.Subscribe(nextItem =>
             {
@@ -76,7 +78,7 @@ private IObservable<Db> ProcessUpdate(IObservable<IDb> dbStream)
 
                 observer.OnNext((Db)nextItem);
                 prev = nextItem;
-            }, observer.OnError, observer.OnCompleted);
+            }, observer.OnCompleted);
         });
     }
 
@@ -134,7 +136,7 @@ public IObservable<IDb> Revisions
         {
             if (_dbStream == default!)
                 ThrowNullDb();
-            return _dbStream!;
+            return ObservableExtensions.AsSystemObservable(_dbStream!);
         }
     }
 
@@ -198,7 +200,7 @@ private void Bootstrap()
             AddMissingAttributes(_declaredAttributes.Values);
 
             _dbStreamDisposable = ProcessUpdate(_store.TxLog)
-                .Subscribe(_dbStream);
+                .Subscribe(itm => _dbStream.OnNext(itm));
         }
         catch (Exception ex)
         {

src/NexusMods.MnemonicDB/NexusMods.MnemonicDB.csproj

@@ -7,6 +7,7 @@
         <PackageReference Include="DynamicData" Version="8.4.1"/>
         <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="8.0.1"/>
         <PackageReference Include="Microsoft.Extensions.ObjectPool" Version="8.0.6" />
+        <PackageReference Include="R3" Version="1.2.8" />
         <PackageReference Include="Reloaded.Memory" Version="9.4.1"/>
         <PackageReference Include="RocksDB" Version="8.11.3.46984" />
         <PackageReference Include="System.Reactive" Version="6.0.1" />

src/NexusMods.MnemonicDB/Storage/DatomStore.cs

@@ -3,7 +3,6 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
-using System.Reactive.Subjects;
 using System.Runtime.InteropServices;
 using System.Threading;
 using System.Threading.Tasks;
@@ -18,6 +17,7 @@
 using NexusMods.MnemonicDB.Abstractions.TxFunctions;
 using NexusMods.MnemonicDB.Storage.Abstractions;
 using NexusMods.MnemonicDB.Storage.DatomStorageStructures;
+using R3;
 using Reloaded.Memory.Extensions;
 
 namespace NexusMods.MnemonicDB.Storage;
@@ -158,7 +158,7 @@ public DatomStore(ILogger<DatomStore> logger, AttributeRegistry registry, DatomS
     }
 
     /// <inheritdoc />
-    public IObservable<IDb> TxLog
+    public Observable<IDb> TxLog
     {
         get
         {