Fixed how partial items are scheduled

Directory.Build.targets

@@ -1,52 +1,53 @@
-<?xml version="1.0" encoding="utf-8"?>
-<Project>
-  <PropertyGroup>
-    <CompilerGeneratedFilesOutputPath>$(BaseIntermediateOutputPath)\GeneratedFiles</CompilerGeneratedFilesOutputPath>
-  </PropertyGroup>
-  <ItemGroup>
-    <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
-    <PackageReference Include="Rocket.Surgery.MSBuild.CI" Version="1.1.0" PrivateAssets="All" />
-    <PackageReference Include="Rocket.Surgery.MSBuild.Metadata" Version="1.1.0" PrivateAssets="All" />
-    <PackageReference Include="Rocket.Surgery.MSBuild.SourceLink" Version="1.1.0" PrivateAssets="All" />
-    <PackageReference Include="Microsoft.NETFramework.ReferenceAssemblies" Version="1.0.0" PrivateAssets="All" />
-  </ItemGroup>
-  <ItemGroup>
-    <PackageReference Update="GitVersion.Tool" Version="5.6.6" />
-    <PackageReference Update="JetBrains.ReSharper.CommandLineTools" Version="2020.3.3" />
-    <PackageReference Update="ReportGenerator" Version="4.8.6" />
-    <PackageReference Update="Rocket.Surgery.Nuke" Version="0.14.3" />
-  </ItemGroup>
-  <ItemGroup>
-    <PackageReference Update="Microsoft.Extensions.Logging" Version="2.0.0" />
-    <PackageReference Update="Microsoft.Extensions.Logging.Debug" Version="2.0.0" />
-    <PackageReference Update="Microsoft.Extensions.Configuration" Version="2.0.0" />
-    <PackageReference Update="Microsoft.Extensions.Configuration.Binder" Version="2.0.0" />
-    <PackageReference Update="Microsoft.Extensions.Options" Version="2.0.0" />
-    <PackageReference Update="Microsoft.Extensions.Options.ConfigurationExtensions" Version="2.0.0" />
-    <PackageReference Update="Microsoft.Extensions.DependencyInjection" Version="2.0.0" />
-    <PackageReference Update="Newtonsoft.Json" Version="11.0.2" />
-    <PackageReference Update="Microsoft.NET.Test.Sdk" Version="16.9.1" />
-    <PackageReference Update="xunit.runner.visualstudio" Version="2.4.3" />
-    <PackageReference Update="xunit" Version="2.4.1" />
-    <PackageReference Update="FluentAssertions" Version="5.10.3" />
-    <PackageReference Update="NSubstitute" Version="4.2.2" />
-    <PackageReference Update="Serilog.Extensions.Logging" Version="2.0.2" />
-    <PackageReference Update="Serilog.Sinks.Observable" Version="2.0.2" />
-    <PackageReference Update="Serilog.Sinks.XUnit" Version="2.0.4" />
-    <PackageReference Update="XunitXml.TestLogger" Version="3.0.62" />
-    <PackageReference Update="coverlet.collector" Version="3.0.3" />
-    <PackageReference Update="coverlet.msbuild" Version="3.0.3" />
-    <PackageReference Update="System.Reactive" Version="4.4.1" />
-    <PackageReference Update="System.Collections.Immutable" Version="1.7.1" />
-    <PackageReference Update="Microsoft.Reactive.Testing" Version="4.4.1" />
-    <PackageReference Update="MediatR" Version="8.1.0" />
-    <PackageReference Update="Bogus" Version="33.0.2" />
-    <PackageReference Update="Snapper" Version="2.3.0" />
-    <PackageReference Update="Xunit.SkippableFact" Version="1.4.13" />
-    <PackageReference Update="System.IO.Pipelines" Version="4.7.3" />
-    <PackageReference Update="Nerdbank.Streams" Version="2.6.81" />
-    <PackageReference Update="Microsoft.CodeAnalysis.CSharp.Workspaces" Version="3.8.0" />
-    <PackageReference Update="Microsoft.CodeAnalysis.Analyzers" Version="3.3.2" />
-    <PackageReference Update="DryIoc.Internal" Version="4.7.3" />
-  </ItemGroup>
-</Project>
+<?xml version="1.0" encoding="utf-8"?>
+<Project>
+    <PropertyGroup>
+        <CompilerGeneratedFilesOutputPath>$(BaseIntermediateOutputPath)\GeneratedFiles</CompilerGeneratedFilesOutputPath>
+    </PropertyGroup>
+    <ItemGroup>
+        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All"/>
+        <PackageReference Include="Rocket.Surgery.MSBuild.CI" Version="1.1.0" PrivateAssets="All"/>
+        <PackageReference Include="Rocket.Surgery.MSBuild.Metadata" Version="1.1.0" PrivateAssets="All"/>
+        <PackageReference Include="Rocket.Surgery.MSBuild.SourceLink" Version="1.1.0" PrivateAssets="All"/>
+        <PackageReference Include="Microsoft.NETFramework.ReferenceAssemblies" Version="1.0.0" PrivateAssets="All"/>
+    </ItemGroup>
+    <ItemGroup>
+        <PackageReference Update="GitVersion.Tool" Version="5.6.6"/>
+        <PackageReference Update="JetBrains.ReSharper.CommandLineTools" Version="2020.3.3"/>
+        <PackageReference Update="ReportGenerator" Version="4.8.6"/>
+        <PackageReference Update="Rocket.Surgery.Nuke" Version="0.14.3"/>
+    </ItemGroup>
+    <ItemGroup>
+        <PackageReference Update="Microsoft.Extensions.Logging" Version="2.0.0"/>
+        <PackageReference Update="Microsoft.Extensions.Logging.Debug" Version="2.0.0"/>
+        <PackageReference Update="Microsoft.Extensions.Configuration" Version="2.0.0"/>
+        <PackageReference Update="Microsoft.Extensions.Configuration.Binder" Version="2.0.0"/>
+        <PackageReference Update="Microsoft.Extensions.Options" Version="2.0.0"/>
+        <PackageReference Update="Microsoft.Extensions.Options.ConfigurationExtensions" Version="2.0.0"/>
+        <PackageReference Update="Microsoft.Extensions.DependencyInjection" Version="2.0.0"/>
+        <PackageReference Update="Newtonsoft.Json" Version="11.0.2"/>
+        <PackageReference Update="Microsoft.NET.Test.Sdk" Version="16.9.1"/>
+        <PackageReference Update="xunit.runner.visualstudio" Version="2.4.3"/>
+        <PackageReference Update="xunit" Version="2.4.1"/>
+        <PackageReference Update="FluentAssertions" Version="5.10.3"/>
+        <PackageReference Update="NSubstitute" Version="4.2.2"/>
+        <PackageReference Update="Serilog.Extensions.Logging" Version="2.0.2"/>
+        <PackageReference Update="Serilog.Sinks.Observable" Version="2.0.2"/>
+        <PackageReference Update="Serilog.Sinks.XUnit" Version="2.0.4"/>
+        <PackageReference Update="XunitXml.TestLogger" Version="3.0.62"/>
+        <PackageReference Update="coverlet.collector" Version="3.0.3"/>
+        <PackageReference Update="coverlet.msbuild" Version="3.0.3"/>
+        <PackageReference Update="System.Reactive" Version="4.4.1"/>
+        <PackageReference Update="System.Collections.Immutable" Version="1.7.1"/>
+        <PackageReference Update="System.Threading.Channels" Version="4.7.1"/>
+        <PackageReference Update="Microsoft.Reactive.Testing" Version="4.4.1"/>
+        <PackageReference Update="MediatR" Version="8.1.0"/>
+        <PackageReference Update="Bogus" Version="33.0.2"/>
+        <PackageReference Update="Snapper" Version="2.3.0"/>
+        <PackageReference Update="Xunit.SkippableFact" Version="1.4.13"/>
+        <PackageReference Update="System.IO.Pipelines" Version="4.7.3"/>
+        <PackageReference Update="Nerdbank.Streams" Version="2.6.81"/>
+        <PackageReference Update="Microsoft.CodeAnalysis.CSharp.Workspaces" Version="3.8.0"/>
+        <PackageReference Update="Microsoft.CodeAnalysis.Analyzers" Version="3.3.2"/>
+        <PackageReference Update="DryIoc.Internal" Version="4.7.3"/>
+    </ItemGroup>
+</Project>

src/Client/LanguageClient.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
 using System.Linq;
+using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
@@ -48,6 +49,7 @@ public class LanguageClient : JsonRpcServerBase, ILanguageClient
         private readonly IEnumerable<IOnLanguageClientInitialized> _initializedHandlers;
         private readonly LspSerializer _serializer;
         private readonly InstanceHasStarted _instanceHasStarted;
+        private readonly IScheduler _scheduler;
         private readonly IResponseRouter _responseRouter;
         private readonly ISubject<InitializeResult> _initializeComplete = new AsyncSubject<InitializeResult>();
         private readonly CompositeDisposable _disposable = new CompositeDisposable();
@@ -150,7 +152,8 @@ internal LanguageClient(
             IEnumerable<OnLanguageClientInitializedDelegate> initializedDelegates,
             IEnumerable<IOnLanguageClientInitialized> initializedHandlers,
             LspSerializer serializer,
-            InstanceHasStarted instanceHasStarted
+            InstanceHasStarted instanceHasStarted,
+            IScheduler scheduler
         ) : base(handlerCollection, responseRouter)
         {
             _connection = connection;
@@ -179,6 +182,7 @@ InstanceHasStarted instanceHasStarted
             _initializedHandlers = initializedHandlers;
             _serializer = serializer;
             _instanceHasStarted = instanceHasStarted;
+            _scheduler = scheduler;
             _concurrency = options.Value.Concurrency;
 
             // We need to at least create Window here in case any handler does loggin in their constructor
@@ -262,6 +266,7 @@ await LanguageProtocolEventingHelper.Run(
                 _initializeHandlers.Union(_collection.Select(z => z.Handler).OfType<IOnLanguageClientInitialize>()),
                 (handler, ct) => handler.OnInitialize(this, @params, ct),
                 _concurrency,
+                _scheduler,
                 token
             ).ConfigureAwait(false);
 
@@ -281,6 +286,7 @@ await LanguageProtocolEventingHelper.Run(
                 _initializedHandlers.Union(_collection.Select(z => z.Handler).OfType<IOnLanguageClientInitialized>()),
                 (handler, ct) => handler.OnInitialized(this, @params, serverParams, ct),
                 _concurrency,
+                _scheduler,
                 token
             ).ConfigureAwait(false);
 
@@ -299,6 +305,7 @@ await LanguageProtocolEventingHelper.Run(
                 _startedHandlers.Union(_collection.Select(z => z.Handler).OfType<IOnLanguageClientStarted>()),
                 (handler, ct) => handler.OnStarted(this, ct),
                 _concurrency,
+                _scheduler,
                 token
             ).ConfigureAwait(false);
 
@@ -395,7 +402,7 @@ private Supports<T> UseOrTryAndFindCapability<T>(Supports<T> supports) where T :
         bool IResponseRouter.TryGetRequest(long id, [NotNullWhen(true)] out string method, [NotNullWhen(true)] out TaskCompletionSource<JToken> pendingTask) =>
             _responseRouter.TryGetRequest(id, out method, out pendingTask);
 
-        public Task<InitializeResult> WasStarted => _initializeComplete.ToTask();
+        public Task<InitializeResult> WasStarted => _initializeComplete.ToTask(_scheduler);
 
         public void Dispose()
         {

src/Client/LanguageClientOptionsExtensions.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Reactive.Concurrency;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
@@ -87,6 +88,54 @@ public static LanguageClientOptions WithClientCapabilities(this LanguageClientOp
             return options;
         }
 
+        /// <summary>
+        /// Sets both input and output schedulers to the same scheduler
+        /// </summary>
+        /// <param name="options"></param>
+        /// <param name="inputScheduler"></param>
+        /// <returns></returns>
+        public static LanguageClientOptions WithScheduler(this LanguageClientOptions options, IScheduler inputScheduler)
+        {
+            options.InputScheduler = options.OutputScheduler = options.DefaultScheduler = inputScheduler;
+            return options;
+        }
+
+        /// <summary>
+        /// Sets the scheduler used during reading input
+        /// </summary>
+        /// <param name="options"></param>
+        /// <param name="inputScheduler"></param>
+        /// <returns></returns>
+        public static LanguageClientOptions WithInputScheduler(this LanguageClientOptions options, IScheduler inputScheduler)
+        {
+            options.InputScheduler = inputScheduler;
+            return options;
+        }
+
+        /// <summary>
+        /// Sets the default scheduler to be used when scheduling other tasks
+        /// </summary>
+        /// <param name="options"></param>
+        /// <param name="defaultScheduler"></param>
+        /// <returns></returns>
+        public static LanguageClientOptions WithDefaultScheduler(this LanguageClientOptions options, IScheduler defaultScheduler)
+        {
+            options.DefaultScheduler = defaultScheduler;
+            return options;
+        }
+
+        /// <summary>
+        /// Sets the scheduler use during writing output
+        /// </summary>
+        /// <param name="options"></param>
+        /// <param name="outputScheduler"></param>
+        /// <returns></returns>
+        public static LanguageClientOptions WithOutputScheduler(this LanguageClientOptions options, IScheduler outputScheduler)
+        {
+            options.OutputScheduler = outputScheduler;
+            return options;
+        }
+
         public static LanguageClientOptions OnInitialize(this LanguageClientOptions options, OnLanguageClientInitializeDelegate @delegate)
         {
             options.Services.AddSingleton(@delegate);

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/DebugAdapterClient.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 using System.Reactive.Subjects;
 using System.Reactive.Threading.Tasks;
@@ -30,6 +31,7 @@ public class DebugAdapterClient : JsonRpcServerBase, IDebugAdapterClient, IDebug
         private readonly IEnumerable<OnDebugAdapterClientStartedDelegate> _startedDelegates;
         private readonly IEnumerable<IOnDebugAdapterClientStarted> _startedHandlers;
         private readonly InstanceHasStarted _instanceHasStarted;
+        private readonly IScheduler _scheduler;
         private readonly CompositeDisposable _disposable = new CompositeDisposable();
         private readonly Connection _connection;
         private readonly DapReceiver _receiver;
@@ -97,7 +99,8 @@ internal DebugAdapterClient(
             IEnumerable<OnDebugAdapterClientInitializedDelegate> initializedDelegates,
             IEnumerable<IOnDebugAdapterClientInitialized> initializedHandlers,
             IEnumerable<IOnDebugAdapterClientStarted> startedHandlers,
-            InstanceHasStarted instanceHasStarted
+            InstanceHasStarted instanceHasStarted,
+            IScheduler scheduler
         ) : base(collection, responseRouter)
         {
             _settingsBag = settingsBag;
@@ -114,6 +117,7 @@ InstanceHasStarted instanceHasStarted
             _initializedHandlers = initializedHandlers;
             _startedHandlers = startedHandlers;
             _instanceHasStarted = instanceHasStarted;
+            _scheduler = scheduler;
             _concurrency = options.Value.Concurrency;
 
             _disposable.Add(collection.Add(this));
@@ -127,6 +131,7 @@ await DebugAdapterEventingHelper.Run(
                 _initializeHandlers.Union(_collection.Select(z => z.Handler).OfType<IOnDebugAdapterClientInitialize>()),
                 (handler, ct) => handler.OnInitialize(this, ClientSettings, ct),
                 _concurrency,
+                _scheduler,
                 token
             ).ConfigureAwait(false);
 
@@ -145,17 +150,19 @@ await DebugAdapterEventingHelper.Run(
                 _initializedHandlers.Union(_collection.Select(z => z.Handler).OfType<IOnDebugAdapterClientInitialized>()),
                 (handler, ct) => handler.OnInitialized(this, ClientSettings, ServerSettings, ct),
                 _concurrency,
+                _scheduler,
                 token
             ).ConfigureAwait(false);
 
-            await _initializedComplete.ToTask(token);
+            await _initializedComplete.ToTask(token, _scheduler);
 
             await DebugAdapterEventingHelper.Run(
                 _startedDelegates,
                 (handler, ct) => handler(this, ct),
                 _startedHandlers.Union(_collection.Select(z => z.Handler).OfType<IOnDebugAdapterClientStarted>()),
                 (handler, ct) => handler.OnStarted(this, ct),
                 _concurrency,
+                _scheduler,
                 token
             ).ConfigureAwait(false);