Fixed schedulers for replay subjects

Author: david-driscoll

src/Client/LanguageClientRegistrationManager.cs

@@ -2,6 +2,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Concurrency;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using System.Threading;
@@ -26,7 +27,7 @@ internal class LanguageClientRegistrationManager : IRegisterCapabilityHandler, I
         private readonly ILspHandlerTypeDescriptorProvider _handlerTypeDescriptorProvider;
         private readonly ILogger<LanguageClientRegistrationManager> _logger;
         private readonly ConcurrentDictionary<string, Registration> _registrations;
-        private readonly ReplaySubject<IEnumerable<Registration>> _registrationSubject = new ReplaySubject<IEnumerable<Registration>>(1);
+        private readonly ReplaySubject<IEnumerable<Registration>> _registrationSubject = new ReplaySubject<IEnumerable<Registration>>(1, Scheduler.Immediate);
 
         public LanguageClientRegistrationManager(
             ISerializer serializer,

src/Client/LanguageClientWorkspaceFoldersManager.cs

@@ -2,6 +2,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Concurrency;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using System.Threading;
@@ -25,7 +26,7 @@ public LanguageClientWorkspaceFoldersManager(IWorkspaceLanguageClient client, IE
         {
             _client = client;
             _workspaceFolders = new ConcurrentDictionary<DocumentUri, WorkspaceFolder>(DocumentUri.Comparer);
-            _workspaceFoldersSubject = new ReplaySubject<IEnumerable<WorkspaceFolder>>(1);
+            _workspaceFoldersSubject = new ReplaySubject<IEnumerable<WorkspaceFolder>>(1, Scheduler.Immediate);
 
             foreach (var folder in workspaceFolders)
             {

src/Dap.Client/ProgressObservable.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 using System.Reactive.Subjects;
 using OmniSharp.Extensions.DebugAdapter.Protocol;
@@ -14,7 +15,7 @@ internal class ProgressObservable : IProgressObservable, IObserver<ProgressEvent
 
         public ProgressObservable(ProgressToken token)
         {
-            _dataSubject = new ReplaySubject<ProgressEvent>(1);
+            _dataSubject = new ReplaySubject<ProgressEvent>(1, Scheduler.Immediate);
             _disposable = new CompositeDisposable { Disposable.Create(_dataSubject.OnCompleted) };
 
             ProgressToken = token;

src/JsonRpc/OutputHandler.cs

@@ -39,7 +39,7 @@ ILogger<OutputHandler> logger
             _outputFilters = outputFilters.ToArray();
             _logger = logger;
             _queue = new Subject<object>();
-            _delayedQueue = new ReplaySubject<object>();
+            _delayedQueue = new ReplaySubject<object>(Scheduler.Immediate);
             _outputIsFinished = new TaskCompletionSource<object?>();
 
             _disposable = new CompositeDisposable {

src/JsonRpc/ProcessScheduler.cs

@@ -34,7 +34,7 @@ IScheduler scheduler
 
                     var observableQueue =
                         new BehaviorSubject<(RequestProcessType type, ReplaySubject<IObservable<Unit>> observer, Subject<Unit>? contentModifiedSource)>(
-                            ( RequestProcessType.Serial, new ReplaySubject<IObservable<Unit>>(int.MaxValue), supportContentModified ? new Subject<Unit>() : null )
+                            ( RequestProcessType.Serial, new ReplaySubject<IObservable<Unit>>(int.MaxValue, Scheduler.Immediate), supportContentModified ? new Subject<Unit>() : null )
                         );
 
                     cd.Add(
@@ -52,7 +52,7 @@ IScheduler scheduler
 
                                     logger.LogDebug("Completing existing request process type {Type}", observableQueue.Value.type);
                                     observableQueue.Value.observer.OnCompleted();
-                                    observableQueue.OnNext(( item.type, new ReplaySubject<IObservable<Unit>>(int.MaxValue), supportContentModified ? new Subject<Unit>() : null ));
+                                    observableQueue.OnNext(( item.type, new ReplaySubject<IObservable<Unit>>(int.MaxValue, Scheduler.Immediate), supportContentModified ? new Subject<Unit>() : null ));
                                 }
 
                                 logger.LogDebug("Queueing {Type}:{Name} request for processing", item.type, item.name);

src/Protocol/Progress/PartialItemsRequestProgressObservable.cs

@@ -36,7 +36,7 @@ Action onCompleteAction
         )
         {
             _serializer = serializer;
-            _dataSubject = new ReplaySubject<IEnumerable<TItem>>(int.MaxValue);
+            _dataSubject = new ReplaySubject<IEnumerable<TItem>>(int.MaxValue, Scheduler.Immediate);
             _disposable = new CompositeDisposable() { _dataSubject };
 
             _task = Observable.Create<TResult>(

src/Protocol/Progress/ProgressObservable.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
@@ -16,7 +17,7 @@ internal class ProgressObservable<T> : IProgressObservable<T>, IObserver<JToken>
         public ProgressObservable(ProgressToken token, Func<JToken, T> factory, Action disposal)
         {
             _factory = factory;
-            _dataSubject = new ReplaySubject<JToken>(1);
+            _dataSubject = new ReplaySubject<JToken>(1, Scheduler.Immediate);
             _disposable = new CompositeDisposable { Disposable.Create(_dataSubject.OnCompleted), Disposable.Create(disposal) };
 
             ProgressToken = token;

src/Server/LanguageServerWorkspaceFolderManager.cs

@@ -2,6 +2,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Concurrency;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using System.Threading;
@@ -27,7 +28,7 @@ public LanguageServerWorkspaceFolderManager(IWorkspaceLanguageServer server)
         {
             _server = server;
             _workspaceFolders = new ConcurrentDictionary<DocumentUri, WorkspaceFolder>(DocumentUri.Comparer);
-            _workspaceFoldersSubject = new ReplaySubject<IEnumerable<WorkspaceFolder>>(1);
+            _workspaceFoldersSubject = new ReplaySubject<IEnumerable<WorkspaceFolder>>(1, Scheduler.Immediate);
             _workspaceFoldersChangedSubject = new Subject<WorkspaceFolderChange>();
         }
 

test/Lsp.Tests/Integration/DynamicRegistrationTests.cs

@@ -119,6 +119,8 @@ await TestHelper.DelayUntil(
                     registrations => registrations.Any(registration => SelectorMatches(registration, x => x.HasLanguage && x.Language == "vb")),
                     CancellationToken
                 );
+
+                await Task.Delay(200);
                 disposable.Dispose();
 
 
@@ -127,6 +129,7 @@ await TestHelper.DelayUntil(
                     registrations => !registrations.Any(registration => SelectorMatches(registration, x => x.HasLanguage && x.Language == "vb")),
                     CancellationToken
                 );
+                await Task.Delay(200);
 
                 client.RegistrationManager.CurrentRegistrations.Should().NotContain(
                     x =>

test/Lsp.Tests/Integration/LanguageServerConfigurationTests.cs

@@ -254,15 +254,15 @@ public async Task Should_Support_Options_Monitor()
             // IOptionsMonitor<> is registered as a singleton, so this will update
             options.CurrentValue.Host.Should().Be("localhost");
             options.CurrentValue.Port.Should().Be(443);
-            sub.Received(1).Invoke(Arg.Any<BinderSourceUrl>());
+            sub.Received(Quantity.AtLeastOne()).Invoke(Arg.Any<BinderSourceUrl>());
 
             configuration.Update("mysection", new Dictionary<string, string> { ["host"] = "127.0.0.1", ["port"] = "80" });
             await options.WaitForChange(CancellationToken);
             await SettleNext();
 
             options.CurrentValue.Host.Should().Be("127.0.0.1");
             options.CurrentValue.Port.Should().Be(80);
-            sub.Received(2).Invoke(Arg.Any<BinderSourceUrl>());
+            sub.Received(Quantity.Within(2, int.MaxValue)).Invoke(Arg.Any<BinderSourceUrl>());
         }
 
         class BinderSourceUrl