Skip to content

Commit 74660a7

Browse files
david-driscolldavid-driscoll
committed
Fixed schedulers for replay subjects
1 parent 0dae323 commit 74660a7

24 files changed

+459
-77
lines changed

‎src/Client/LanguageClientRegistrationManager.cs

+2-1
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
using System.Collections.Concurrent;
33
using System.Collections.Generic;
44
using System.Linq;
5+
using System.Reactive.Concurrency;
56
using System.Reactive.Linq;
67
using System.Reactive.Subjects;
78
using System.Threading;
@@ -26,7 +27,7 @@ internal class LanguageClientRegistrationManager : IRegisterCapabilityHandler, I
2627
private readonly ILspHandlerTypeDescriptorProvider _handlerTypeDescriptorProvider;
2728
private readonly ILogger<LanguageClientRegistrationManager> _logger;
2829
private readonly ConcurrentDictionary<string, Registration> _registrations;
29-
private readonly ReplaySubject<IEnumerable<Registration>> _registrationSubject = new ReplaySubject<IEnumerable<Registration>>(1);
30+
private readonly ReplaySubject<IEnumerable<Registration>> _registrationSubject = new ReplaySubject<IEnumerable<Registration>>(1, Scheduler.Immediate);
3031

3132
public LanguageClientRegistrationManager(
3233
ISerializer serializer,

‎src/Client/LanguageClientWorkspaceFoldersManager.cs

+2-1
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
using System.Collections.Concurrent;
33
using System.Collections.Generic;
44
using System.Linq;
5+
using System.Reactive.Concurrency;
56
using System.Reactive.Linq;
67
using System.Reactive.Subjects;
78
using System.Threading;
@@ -25,7 +26,7 @@ public LanguageClientWorkspaceFoldersManager(IWorkspaceLanguageClient client, IE
2526
{
2627
_client = client;
2728
_workspaceFolders = new ConcurrentDictionary<DocumentUri, WorkspaceFolder>(DocumentUri.Comparer);
28-
_workspaceFoldersSubject = new ReplaySubject<IEnumerable<WorkspaceFolder>>(1);
29+
_workspaceFoldersSubject = new ReplaySubject<IEnumerable<WorkspaceFolder>>(1, Scheduler.Immediate);
2930

3031
foreach (var folder in workspaceFolders)
3132
{

‎src/Dap.Client/ProgressObservable.cs

+2-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
using System;
2+
using System.Reactive.Concurrency;
23
using System.Reactive.Disposables;
34
using System.Reactive.Subjects;
45
using OmniSharp.Extensions.DebugAdapter.Protocol;
@@ -14,7 +15,7 @@ internal class ProgressObservable : IProgressObservable, IObserver<ProgressEvent
1415

1516
public ProgressObservable(ProgressToken token)
1617
{
17-
_dataSubject = new ReplaySubject<ProgressEvent>(1);
18+
_dataSubject = new ReplaySubject<ProgressEvent>(1, Scheduler.Immediate);
1819
_disposable = new CompositeDisposable { Disposable.Create(_dataSubject.OnCompleted) };
1920

2021
ProgressToken = token;

‎src/JsonRpc/ProcessScheduler.cs

+2-2
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ IScheduler scheduler
3434

3535
var observableQueue =
3636
new BehaviorSubject<(RequestProcessType type, ReplaySubject<IObservable<Unit>> observer, Subject<Unit>? contentModifiedSource)>(
37-
( RequestProcessType.Serial, new ReplaySubject<IObservable<Unit>>(int.MaxValue), supportContentModified ? new Subject<Unit>() : null )
37+
( RequestProcessType.Serial, new ReplaySubject<IObservable<Unit>>(int.MaxValue, Scheduler.Immediate), supportContentModified ? new Subject<Unit>() : null )
3838
);
3939

4040
cd.Add(
@@ -52,7 +52,7 @@ IScheduler scheduler
5252

5353
logger.LogDebug("Completing existing request process type {Type}", observableQueue.Value.type);
5454
observableQueue.Value.observer.OnCompleted();
55-
observableQueue.OnNext(( item.type, new ReplaySubject<IObservable<Unit>>(int.MaxValue), supportContentModified ? new Subject<Unit>() : null ));
55+
observableQueue.OnNext(( item.type, new ReplaySubject<IObservable<Unit>>(int.MaxValue, Scheduler.Immediate), supportContentModified ? new Subject<Unit>() : null ));
5656
}
5757

5858
logger.LogDebug("Queueing {Type}:{Name} request for processing", item.type, item.name);

‎src/Protocol/Progress/PartialItemsRequestProgressObservable.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ Action onCompleteAction
3636
)
3737
{
3838
_serializer = serializer;
39-
_dataSubject = new ReplaySubject<IEnumerable<TItem>>(int.MaxValue);
39+
_dataSubject = new ReplaySubject<IEnumerable<TItem>>(int.MaxValue, Scheduler.Immediate);
4040
_disposable = new CompositeDisposable() { _dataSubject };
4141

4242
_task = Observable.Create<TResult>(

‎src/Protocol/Progress/ProgressObservable.cs

+2-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
using System;
2+
using System.Reactive.Concurrency;
23
using System.Reactive.Disposables;
34
using System.Reactive.Linq;
45
using System.Reactive.Subjects;
@@ -16,7 +17,7 @@ internal class ProgressObservable<T> : IProgressObservable<T>, IObserver<JToken>
1617
public ProgressObservable(ProgressToken token, Func<JToken, T> factory, Action disposal)
1718
{
1819
_factory = factory;
19-
_dataSubject = new ReplaySubject<JToken>(1);
20+
_dataSubject = new ReplaySubject<JToken>(1, Scheduler.Immediate);
2021
_disposable = new CompositeDisposable { Disposable.Create(_dataSubject.OnCompleted), Disposable.Create(disposal) };
2122

2223
ProgressToken = token;

‎src/Protocol/Progress/RequestProgressObservable.cs

+7-6
Original file line numberDiff line numberDiff line change
@@ -41,12 +41,13 @@ Action onCompleteAction
4141
observer => new CompositeDisposable() {
4242
_dataSubject
4343
.ForkJoin(
44-
requestResult.Do(
45-
_ => {
46-
if (_receivedPartialData) return;
47-
_dataSubject.OnNext(reverseFactory(_));
48-
}, OnError, OnCompleted
49-
), factory
44+
requestResult
45+
.Do(
46+
_ => _dataSubject.OnNext(reverseFactory(_)),
47+
_dataSubject.OnError,
48+
_dataSubject.OnCompleted
49+
),
50+
factory
5051
)
5152
.Subscribe(observer),
5253
Disposable.Create(onCompleteAction)

‎src/Server/LanguageServerWorkspaceFolderManager.cs

+2-1
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
using System.Collections.Concurrent;
33
using System.Collections.Generic;
44
using System.Linq;
5+
using System.Reactive.Concurrency;
56
using System.Reactive.Linq;
67
using System.Reactive.Subjects;
78
using System.Threading;
@@ -27,7 +28,7 @@ public LanguageServerWorkspaceFolderManager(IWorkspaceLanguageServer server)
2728
{
2829
_server = server;
2930
_workspaceFolders = new ConcurrentDictionary<DocumentUri, WorkspaceFolder>(DocumentUri.Comparer);
30-
_workspaceFoldersSubject = new ReplaySubject<IEnumerable<WorkspaceFolder>>(1);
31+
_workspaceFoldersSubject = new ReplaySubject<IEnumerable<WorkspaceFolder>>(1, Scheduler.Immediate);
3132
_workspaceFoldersChangedSubject = new Subject<WorkspaceFolderChange>();
3233
}
3334

‎test/Lsp.Tests/Integration/DynamicRegistrationTests.cs

+3
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,8 @@ await TestHelper.DelayUntil(
119119
registrations => registrations.Any(registration => SelectorMatches(registration, x => x.HasLanguage && x.Language == "vb")),
120120
CancellationToken
121121
);
122+
123+
await Task.Delay(200);
122124
disposable.Dispose();
123125

124126

@@ -127,6 +129,7 @@ await TestHelper.DelayUntil(
127129
registrations => !registrations.Any(registration => SelectorMatches(registration, x => x.HasLanguage && x.Language == "vb")),
128130
CancellationToken
129131
);
132+
await Task.Delay(200);
130133

131134
client.RegistrationManager.CurrentRegistrations.Should().NotContain(
132135
x =>

‎test/Lsp.Tests/Integration/LanguageServerConfigurationTests.cs

+2-2
Original file line numberDiff line numberDiff line change
@@ -254,15 +254,15 @@ public async Task Should_Support_Options_Monitor()
254254
// IOptionsMonitor<> is registered as a singleton, so this will update
255255
options.CurrentValue.Host.Should().Be("localhost");
256256
options.CurrentValue.Port.Should().Be(443);
257-
sub.Received(1).Invoke(Arg.Any<BinderSourceUrl>());
257+
sub.Received(Quantity.AtLeastOne()).Invoke(Arg.Any<BinderSourceUrl>());
258258

259259
configuration.Update("mysection", new Dictionary<string, string> { ["host"] = "127.0.0.1", ["port"] = "80" });
260260
await options.WaitForChange(CancellationToken);
261261
await SettleNext();
262262

263263
options.CurrentValue.Host.Should().Be("127.0.0.1");
264264
options.CurrentValue.Port.Should().Be(80);
265-
sub.Received(2).Invoke(Arg.Any<BinderSourceUrl>());
265+
sub.Received(Quantity.Within(2, int.MaxValue)).Invoke(Arg.Any<BinderSourceUrl>());
266266
}
267267

268268
class BinderSourceUrl

‎test/Lsp.Tests/Integration/PartialItemTests.cs

+3-3
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ public Delegates(ITestOutputHelper testOutputHelper, LanguageProtocolFixture<Def
2424
{
2525
}
2626

27-
[Fact]
27+
[Fact]//[RetryFact]
2828
public async Task Should_Behave_Like_A_Task()
2929
{
3030
var result = await Client.TextDocument.RequestSemanticTokens(
@@ -34,7 +34,7 @@ public async Task Should_Behave_Like_A_Task()
3434
result!.Data.Should().HaveCount(3);
3535
}
3636

37-
[Fact]
37+
[Fact]//[RetryFact]
3838
public async Task Should_Behave_Like_An_Observable()
3939
{
4040
var items = await Client.TextDocument
@@ -53,7 +53,7 @@ public async Task Should_Behave_Like_An_Observable()
5353
items.Select(z => z.Data.Length).Should().ContainInOrder(1, 2, 3);
5454
}
5555

56-
[Fact]
56+
[Fact]//[RetryFact]
5757
public async Task Should_Behave_Like_An_Observable_Without_Progress_Support()
5858
{
5959
var response = await Client.SendRequest(new SemanticTokensParams { TextDocument = new TextDocumentIdentifier(@"c:\test.cs") }, CancellationToken);

‎test/Lsp.Tests/Integration/PartialItemsTests.cs

+4-4
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ public Delegates(ITestOutputHelper testOutputHelper, LanguageProtocolFixture<Def
2929
{
3030
}
3131

32-
[Fact]
32+
[Fact]//[RetryFact]
3333
public async Task Should_Behave_Like_A_Task()
3434
{
3535
var result = await Client.TextDocument.RequestCodeLens(
@@ -42,7 +42,7 @@ public async Task Should_Behave_Like_A_Task()
4242
result.Select(z => z.Command!.Name).Should().ContainInOrder("CodeLens 1", "CodeLens 2", "CodeLens 3");
4343
}
4444

45-
[Fact]
45+
[Fact]//[RetryFact]
4646
public async Task Should_Behave_Like_An_Observable()
4747
{
4848
var items = await Client.TextDocument
@@ -63,7 +63,7 @@ public async Task Should_Behave_Like_An_Observable()
6363
items.Select(z => z.Command!.Name).Should().ContainInOrder("CodeLens 1", "CodeLens 2", "CodeLens 3");
6464
}
6565

66-
[Fact]
66+
[Fact]//[RetryFact]
6767
public async Task Should_Behave_Like_An_Observable_Without_Progress_Support()
6868
{
6969
var response = await Client.SendRequest(
@@ -120,7 +120,7 @@ public Handlers(ITestOutputHelper testOutputHelper, LanguageProtocolFixture<Defa
120120
{
121121
}
122122

123-
[Fact]
123+
[Fact]//[RetryFact]
124124
public async Task Should_Behave_Like_An_Observable_With_WorkDone()
125125
{
126126
var items = new List<CodeLens>();

‎test/Lsp.Tests/Integration/TypedCodeActionTests.cs

+13-13
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ public TypedCodeActionTests(ITestOutputHelper outputHelper) : base(new JsonRpcTe
2828
{
2929
}
3030

31-
[Fact]
31+
[Fact]//[RetryFact]
3232
public async Task Should_Aggregate_With_All_Related_Handlers()
3333
{
3434
var (client, _) = await Initialize(
@@ -157,7 +157,7 @@ public async Task Should_Aggregate_With_All_Related_Handlers()
157157
actions.Length.Should().Be(3);
158158
}
159159

160-
[Fact]
160+
[Fact]//[RetryFact]
161161
public async Task Should_Resolve_With_Data_Capability()
162162
{
163163
var (client, _) = await Initialize(
@@ -203,7 +203,7 @@ public async Task Should_Resolve_With_Data_Capability()
203203
item.CodeAction!.Command!.Name.Should().Be("resolved");
204204
}
205205

206-
[Fact]
206+
[Fact]//[RetryFact]
207207
public async Task Should_Resolve_With_Partial_Data_Capability()
208208
{
209209
var (client, _) = await Initialize(
@@ -248,7 +248,7 @@ public async Task Should_Resolve_With_Partial_Data_Capability()
248248
item.CodeAction!.Command!.Name.Should().Be("resolved");
249249
}
250250

251-
[Fact]
251+
[Fact]//[RetryFact]
252252
public async Task Should_Resolve_With_Data_CancellationToken()
253253
{
254254
var (client, _) = await Initialize(
@@ -294,7 +294,7 @@ public async Task Should_Resolve_With_Data_CancellationToken()
294294
item.CodeAction!.Command!.Name.Should().Be("resolved");
295295
}
296296

297-
[Fact]
297+
[Fact]//[RetryFact]
298298
public async Task Should_Resolve_With_Partial_Data_CancellationToken()
299299
{
300300
var (client, _) = await Initialize(
@@ -339,7 +339,7 @@ public async Task Should_Resolve_With_Partial_Data_CancellationToken()
339339
item.CodeAction!.Command!.Name.Should().Be("resolved");
340340
}
341341

342-
[Fact]
342+
[Fact]//[RetryFact]
343343
public async Task Should_Resolve_With_Data()
344344
{
345345
var (client, _) = await Initialize(
@@ -385,7 +385,7 @@ public async Task Should_Resolve_With_Data()
385385
item.CodeAction!.Command!.Name.Should().Be("resolved");
386386
}
387387

388-
[Fact]
388+
[Fact]//[RetryFact]
389389
public async Task Should_Resolve_With_Partial_Data()
390390
{
391391
var (client, _) = await Initialize(
@@ -431,7 +431,7 @@ public async Task Should_Resolve_With_Partial_Data()
431431
}
432432

433433

434-
[Fact]
434+
[Fact]//[RetryFact]
435435
public async Task Should_Resolve_Capability()
436436
{
437437
var (client, _) = await Initialize(
@@ -467,7 +467,7 @@ public async Task Should_Resolve_Capability()
467467
item.CodeAction!.Command!.Name.Should().Be("resolved");
468468
}
469469

470-
[Fact]
470+
[Fact]//[RetryFact]
471471
public async Task Should_Resolve_Partial_Capability()
472472
{
473473
var (client, _) = await Initialize(
@@ -502,7 +502,7 @@ public async Task Should_Resolve_Partial_Capability()
502502
item.CodeAction!.Command!.Name.Should().Be("resolved");
503503
}
504504

505-
[Fact]
505+
[Fact]//[RetryFact]
506506
public async Task Should_Resolve_CancellationToken()
507507
{
508508
var (client, _) = await Initialize(
@@ -538,7 +538,7 @@ public async Task Should_Resolve_CancellationToken()
538538
item.CodeAction!.Command!.Name.Should().Be("resolved");
539539
}
540540

541-
[Fact]
541+
[Fact]//[RetryFact]
542542
public async Task Should_Resolve_Partial_CancellationToken()
543543
{
544544
var (client, _) = await Initialize(
@@ -573,7 +573,7 @@ public async Task Should_Resolve_Partial_CancellationToken()
573573
item.CodeAction!.Command!.Name.Should().Be("resolved");
574574
}
575575

576-
[Fact]
576+
[Fact]//[RetryFact]
577577
public async Task Should_Resolve()
578578
{
579579
var (client, _) = await Initialize(
@@ -609,7 +609,7 @@ public async Task Should_Resolve()
609609
item.CodeAction!.Command!.Name.Should().Be("resolved");
610610
}
611611

612-
[Fact]
612+
[Fact]//[RetryFact]
613613
public async Task Should_Resolve_Partial()
614614
{
615615
var (client, _) = await Initialize(

0 commit comments

Comments
 (0)