Skip to content

Commit bfbaaf3

Browse files
fedeAlteriofedeAlterio
authored
Throttle + Remove delay from tests (#10)
1 parent 7067c81 commit bfbaaf3

14 files changed

Lines changed: 604 additions & 34 deletions

‎README.md‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -183,8 +183,11 @@ Transform and compose observable streams:
183183
- `Do` - Perform side effects
184184
- `Wrap` - Wrap observer calls
185185

186-
#### Timing (Throttle in classic Rx.NET)
187-
- `Debounce` - Suppress values followed by another value within a time window; only emit after a quiet period
186+
#### Timing
187+
- `Debounce` - Suppress values followed by another value within a time window; only emit after a quiet period (Throttle in classic Rx.NET)
188+
- `ThrottleFirst` - Emit the first value of each time window and drop the rest
189+
- `ThrottleLast` - Emit only the latest value of each time window, when the window expires
190+
- `ThrottleFirstLast` - Emit the first value of each time window immediately, then the latest value when the window expires
188191

189192
#### Concurrency & Scheduling
190193
- `ObserveOn` - Control execution context for downstream operators

‎src/R3Async.Tests/Factories/CreateTest.cs‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -95,7 +95,6 @@ public async Task ErrorResumeTest()
9595

9696
var exception = await tcs.Task;
9797
exception.ShouldBe(expectedException);
98-
await Task.Delay(100);
9998
results.ShouldBe(new[] { 1, 2 });
10099
}
101100

@@ -480,7 +479,7 @@ public async Task OnNextCancelledWhenProducerCancelsTokenTest()
480479
onNextStarted.TrySetResult();
481480
try
482481
{
483-
await Task.Delay(Timeout.Infinite, token);
482+
await TestHelpers.WaitForCancellationAsync(token);
484483
}
485484
catch (OperationCanceledException)
486485
{
@@ -521,7 +520,7 @@ public async Task OnNextCancelledWhenSubscriptionDisposedTest()
521520
onNextStarted.TrySetResult();
522521
try
523522
{
524-
await Task.Delay(Timeout.Infinite, token);
523+
await TestHelpers.WaitForCancellationAsync(token);
525524
}
526525
catch (OperationCanceledException)
527526
{
@@ -564,7 +563,7 @@ public async Task OnErrorResumeCancelledWhenProducerCancelsTokenTest()
564563
onErrorResumeStarted.TrySetResult();
565564
try
566565
{
567-
await Task.Delay(Timeout.Infinite, token);
566+
await TestHelpers.WaitForCancellationAsync(token);
568567
}
569568
catch (OperationCanceledException)
570569
{
@@ -607,7 +606,7 @@ public async Task OnErrorResumeCancelledWhenSubscriptionDisposedTest()
607606
onErrorResumeStarted.TrySetResult();
608607
try
609608
{
610-
await Task.Delay(Timeout.Infinite, token);
609+
await TestHelpers.WaitForCancellationAsync(token);
611610
}
612611
catch (OperationCanceledException)
613612
{

‎src/R3Async.Tests/Factories/DeferTest.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -362,7 +362,7 @@ public async Task DeferAsyncFactoryTest()
362362
var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
363363
var deferred = AsyncObservable.Defer(async token =>
364364
{
365-
await Task.Delay(10, token);
365+
await Task.Yield();
366366
return AsyncObservable.Create<int>(async (observer, t) =>
367367
{
368368
await observer.OnNextAsync(42, t);

‎src/R3Async.Tests/Factories/FromAsyncTest.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ public async Task FromAsync_DisposeCancelsFactory()
5656
{
5757
try
5858
{
59-
await Task.Delay(Timeout.Infinite, ct);
59+
await TestHelpers.WaitForCancellationAsync(ct);
6060
return 1;
6161
}
6262
catch (OperationCanceledException)

‎src/R3Async.Tests/Factories/ReturnTest.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ public async Task Return_DisposeCancelsOnNext()
3434
{
3535
try
3636
{
37-
await Task.Delay(Timeout.Infinite, token);
37+
await TestHelpers.WaitForCancellationAsync(token);
3838
}
3939
catch (OperationCanceledException)
4040
{

‎src/R3Async.Tests/Operators/CatchTest.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -120,7 +120,7 @@ public async Task Catch_Dispose_DisposesHandler()
120120
{
121121
_ = Task.Run(async () =>
122122
{
123-
await Task.Delay(1);
123+
await Task.Yield();
124124
await observer.OnCompletedAsync(Result.Failure(sourceFail));
125125
});
126126
return AsyncDisposable.Empty;

‎src/R3Async.Tests/Operators/DebounceTest.cs‎

Lines changed: 7 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -9,31 +9,17 @@ public class DebounceTest
99
{
1010
static readonly TimeSpan DueTime = TimeSpan.FromMilliseconds(100);
1111

12-
static async Task WaitForAsync(Func<bool> condition, TimeSpan? timeout = null)
13-
{
14-
var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(5));
15-
while (!condition())
16-
{
17-
if (DateTime.UtcNow > deadline)
18-
throw new TimeoutException("Condition was not met within the timeout.");
19-
await Task.Delay(10);
20-
}
21-
}
22-
2312
[Fact]
2413
public async Task Debounce_EmitsOnlyAfterQuietPeriod()
2514
{
2615
var timeProvider = new FakeTimeProvider();
2716
var subject = Subject.Create<int>();
2817

2918
var results = new List<int>();
30-
var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
19+
var itemAdded = new SemaphoreSlim(0);
3120

3221
await using var subscription = await subject.Values.Debounce(DueTime, timeProvider).SubscribeAsync(
33-
async (x, token) => { lock (results) results.Add(x); },
34-
async (ex, token) => { },
35-
async result => completedTcs.TrySetResult(result),
36-
CancellationToken.None);
22+
async (x, token) => { lock (results) results.Add(x); itemAdded.Release(); }, CancellationToken.None);
3723

3824
// Burst of values within the due time: only the last should survive.
3925
await subject.OnNextAsync(1, CancellationToken.None);
@@ -48,7 +34,7 @@ public async Task Debounce_EmitsOnlyAfterQuietPeriod()
4834

4935
// Now let the due time elapse without new values.
5036
timeProvider.Advance(DueTime);
51-
await WaitForAsync(() => { lock (results) return results.Count == 1; });
37+
await itemAdded.WaitAsync();
5238

5339
lock (results) results.ShouldBe(new[] { 3 });
5440
}
@@ -60,17 +46,18 @@ public async Task Debounce_EmitsEachValueSeparatedByQuietPeriods()
6046
var subject = Subject.Create<int>();
6147

6248
var results = new List<int>();
49+
var itemAdded = new SemaphoreSlim(0);
6350

6451
await using var subscription = await subject.Values.Debounce(DueTime, timeProvider).SubscribeAsync(
65-
async (x, token) => { lock (results) results.Add(x); }, CancellationToken.None);
52+
async (x, token) => { lock (results) results.Add(x); itemAdded.Release(); }, CancellationToken.None);
6653

6754
await subject.OnNextAsync(1, CancellationToken.None);
6855
timeProvider.Advance(DueTime);
69-
await WaitForAsync(() => { lock (results) return results.Count == 1; });
56+
await itemAdded.WaitAsync();
7057

7158
await subject.OnNextAsync(2, CancellationToken.None);
7259
timeProvider.Advance(DueTime);
73-
await WaitForAsync(() => { lock (results) return results.Count == 2; });
60+
await itemAdded.WaitAsync();
7461

7562
lock (results) results.ShouldBe(new[] { 1, 2 });
7663
}
@@ -140,7 +127,6 @@ public async Task Debounce_DisposeStopsEmission()
140127

141128
// Firing the timer after disposal must not emit anything.
142129
timeProvider.Advance(DueTime);
143-
await Task.Delay(50);
144130

145131
lock (results) results.ShouldBeEmpty();
146132
}

‎src/R3Async.Tests/Operators/SwitchTest.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ public async Task Switch_DisposeStopsCurrentInner()
157157
tcsStarted.TrySetResult();
158158
try
159159
{
160-
await Task.Delay(Timeout.Infinite, token);
160+
await TestHelpers.WaitForCancellationAsync(token);
161161
}
162162
catch (OperationCanceledException)
163163
{

‎src/R3Async.Tests/Operators/ThrottleFirstLastTest.cs‎

Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,137 @@
1+
using Microsoft.Extensions.Time.Testing;
2+
using R3Async.Subjects;
3+
using Shouldly;
4+
#pragma warning disable CS1998
5+
6+
namespace R3Async.Tests.Operators;
7+
8+
public class ThrottleFirstLastTest
9+
{
10+
static readonly TimeSpan DueTime = TimeSpan.FromMilliseconds(100);
11+
12+
[Fact]
13+
public async Task ThrottleFirstLast_EmitsFirstValueThenLastValueOfWindow()
14+
{
15+
var timeProvider = new FakeTimeProvider();
16+
var subject = Subject.Create<int>();
17+
18+
var results = new List<int>();
19+
var itemAdded = new SemaphoreSlim(0);
20+
21+
await using var subscription = await subject.Values.ThrottleFirstLast(DueTime, timeProvider).SubscribeAsync(
22+
async (x, token) => { lock (results) results.Add(x); itemAdded.Release(); }, CancellationToken.None);
23+
24+
// The first value of the window is emitted immediately.
25+
await subject.OnNextAsync(1, CancellationToken.None);
26+
await itemAdded.WaitAsync();
27+
lock (results) results.ShouldBe(new[] { 1 });
28+
29+
// Values arriving inside the window are not emitted yet; only the latest is kept.
30+
timeProvider.Advance(TimeSpan.FromMilliseconds(50));
31+
await subject.OnNextAsync(2, CancellationToken.None);
32+
await subject.OnNextAsync(3, CancellationToken.None);
33+
lock (results) results.ShouldBe(new[] { 1 });
34+
35+
// When the window elapses, the last pending value is emitted.
36+
timeProvider.Advance(DueTime);
37+
await itemAdded.WaitAsync();
38+
lock (results) results.ShouldBe(new[] { 1, 3 });
39+
40+
// The window is now closed: the next value opens a new one and is emitted immediately.
41+
await subject.OnNextAsync(4, CancellationToken.None);
42+
await itemAdded.WaitAsync();
43+
lock (results) results.ShouldBe(new[] { 1, 3, 4 });
44+
}
45+
46+
[Fact]
47+
public async Task ThrottleFirstLast_EmptyWindowDoesNotEmitOnTimerExpiry()
48+
{
49+
var timeProvider = new FakeTimeProvider();
50+
var subject = Subject.Create<int>();
51+
52+
var results = new List<int>();
53+
var itemAdded = new SemaphoreSlim(0);
54+
55+
await using var subscription = await subject.Values.ThrottleFirstLast(DueTime, timeProvider).SubscribeAsync(
56+
async (x, token) => { lock (results) results.Add(x); itemAdded.Release(); }, CancellationToken.None);
57+
58+
// A single value with no followers: nothing extra is emitted when the window closes.
59+
await subject.OnNextAsync(1, CancellationToken.None);
60+
await itemAdded.WaitAsync();
61+
timeProvider.Advance(DueTime);
62+
63+
lock (results) results.ShouldBe(new[] { 1 });
64+
}
65+
66+
[Fact]
67+
public async Task ThrottleFirstLast_FlushesPendingValueOnCompletion()
68+
{
69+
var timeProvider = new FakeTimeProvider();
70+
var subject = Subject.Create<int>();
71+
72+
var results = new List<int>();
73+
var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
74+
75+
await using var subscription = await subject.Values.ThrottleFirstLast(DueTime, timeProvider).SubscribeAsync(
76+
async (x, token) => { lock (results) results.Add(x); },
77+
async (ex, token) => { },
78+
async result => completedTcs.TrySetResult(result),
79+
CancellationToken.None);
80+
81+
await subject.OnNextAsync(1, CancellationToken.None);
82+
await subject.OnNextAsync(2, CancellationToken.None);
83+
// Complete before the window elapses: the pending value is flushed.
84+
await subject.OnCompletedAsync(Result.Success);
85+
86+
var result = await completedTcs.Task;
87+
result.IsSuccess.ShouldBeTrue();
88+
lock (results) results.ShouldBe(new[] { 1, 2 });
89+
}
90+
91+
[Fact]
92+
public async Task ThrottleFirstLast_ErrorDropsPendingValueAndCompletes()
93+
{
94+
var expected = new InvalidOperationException("boom");
95+
var timeProvider = new FakeTimeProvider();
96+
var subject = Subject.Create<int>();
97+
98+
var results = new List<int>();
99+
var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
100+
101+
await using var subscription = await subject.Values.ThrottleFirstLast(DueTime, timeProvider).SubscribeAsync(
102+
async (x, token) => { lock (results) results.Add(x); },
103+
async (ex, token) => { },
104+
async result => completedTcs.TrySetResult(result),
105+
CancellationToken.None);
106+
107+
await subject.OnNextAsync(7, CancellationToken.None);
108+
await subject.OnNextAsync(8, CancellationToken.None);
109+
await subject.OnCompletedAsync(Result.Failure(expected));
110+
111+
var result = await completedTcs.Task;
112+
result.IsFailure.ShouldBeTrue();
113+
result.Exception.ShouldBe(expected);
114+
lock (results) results.ShouldBe(new[] { 7 });
115+
}
116+
117+
[Fact]
118+
public async Task ThrottleFirstLast_DisposeStopsEmission()
119+
{
120+
var timeProvider = new FakeTimeProvider();
121+
var subject = Subject.Create<int>();
122+
123+
var results = new List<int>();
124+
125+
var subscription = await subject.Values.ThrottleFirstLast(DueTime, timeProvider).SubscribeAsync(
126+
async (x, token) => { lock (results) results.Add(x); }, CancellationToken.None);
127+
128+
await subject.OnNextAsync(1, CancellationToken.None);
129+
await subscription.DisposeAsync();
130+
131+
// Values after disposal must not be emitted, even after the window elapses.
132+
timeProvider.Advance(DueTime);
133+
await subject.OnNextAsync(2, CancellationToken.None);
134+
135+
lock (results) results.ShouldBe(new[] { 1 });
136+
}
137+
}

0 commit comments

Comments
 (0)