Skip to content

Commit 40dd682

Browse files
fedeAlteriofedeAlterio
committed
Debounce
1 parent 2dbf348 commit 40dd682

3 files changed

Lines changed: 318 additions & 0 deletions

File tree

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

Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
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 DebounceTest
9+
{
10+
static readonly TimeSpan DueTime = TimeSpan.FromMilliseconds(100);
11+
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+
23+
[Fact]
24+
public async Task Debounce_EmitsOnlyAfterQuietPeriod()
25+
{
26+
var timeProvider = new FakeTimeProvider();
27+
var subject = Subject.Create<int>();
28+
29+
var results = new List<int>();
30+
var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
31+
32+
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);
37+
38+
// Burst of values within the due time: only the last should survive.
39+
await subject.OnNextAsync(1, CancellationToken.None);
40+
timeProvider.Advance(TimeSpan.FromMilliseconds(50));
41+
await subject.OnNextAsync(2, CancellationToken.None);
42+
timeProvider.Advance(TimeSpan.FromMilliseconds(50));
43+
await subject.OnNextAsync(3, CancellationToken.None);
44+
45+
// Not enough silence yet.
46+
timeProvider.Advance(TimeSpan.FromMilliseconds(50));
47+
lock (results) results.ShouldBeEmpty();
48+
49+
// Now let the due time elapse without new values.
50+
timeProvider.Advance(DueTime);
51+
await WaitForAsync(() => { lock (results) return results.Count == 1; });
52+
53+
lock (results) results.ShouldBe(new[] { 3 });
54+
}
55+
56+
[Fact]
57+
public async Task Debounce_EmitsEachValueSeparatedByQuietPeriods()
58+
{
59+
var timeProvider = new FakeTimeProvider();
60+
var subject = Subject.Create<int>();
61+
62+
var results = new List<int>();
63+
64+
await using var subscription = await subject.Values.Debounce(DueTime, timeProvider).SubscribeAsync(
65+
async (x, token) => { lock (results) results.Add(x); }, CancellationToken.None);
66+
67+
await subject.OnNextAsync(1, CancellationToken.None);
68+
timeProvider.Advance(DueTime);
69+
await WaitForAsync(() => { lock (results) return results.Count == 1; });
70+
71+
await subject.OnNextAsync(2, CancellationToken.None);
72+
timeProvider.Advance(DueTime);
73+
await WaitForAsync(() => { lock (results) return results.Count == 2; });
74+
75+
lock (results) results.ShouldBe(new[] { 1, 2 });
76+
}
77+
78+
[Fact]
79+
public async Task Debounce_FlushesPendingValueOnCompletion()
80+
{
81+
var timeProvider = new FakeTimeProvider();
82+
var subject = Subject.Create<int>();
83+
84+
var results = new List<int>();
85+
var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
86+
87+
await using var subscription = await subject.Values.Debounce(DueTime, timeProvider).SubscribeAsync(
88+
async (x, token) => { lock (results) results.Add(x); },
89+
async (ex, token) => { },
90+
async result => completedTcs.TrySetResult(result),
91+
CancellationToken.None);
92+
93+
await subject.OnNextAsync(42, CancellationToken.None);
94+
// Complete before the due time elapses: the pending value is flushed.
95+
await subject.OnCompletedAsync(Result.Success);
96+
97+
var result = await completedTcs.Task;
98+
result.IsSuccess.ShouldBeTrue();
99+
lock (results) results.ShouldBe(new[] { 42 });
100+
}
101+
102+
[Fact]
103+
public async Task Debounce_ErrorDropsPendingValueAndCompletes()
104+
{
105+
var expected = new InvalidOperationException("boom");
106+
var timeProvider = new FakeTimeProvider();
107+
var subject = Subject.Create<int>();
108+
109+
var results = new List<int>();
110+
var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
111+
112+
await using var subscription = await subject.Values.Debounce(DueTime, timeProvider).SubscribeAsync(
113+
async (x, token) => { lock (results) results.Add(x); },
114+
async (ex, token) => { },
115+
async result => completedTcs.TrySetResult(result),
116+
CancellationToken.None);
117+
118+
await subject.OnNextAsync(7, CancellationToken.None);
119+
await subject.OnCompletedAsync(Result.Failure(expected));
120+
121+
var result = await completedTcs.Task;
122+
result.IsFailure.ShouldBeTrue();
123+
result.Exception.ShouldBe(expected);
124+
lock (results) results.ShouldBeEmpty();
125+
}
126+
127+
[Fact]
128+
public async Task Debounce_DisposeStopsEmission()
129+
{
130+
var timeProvider = new FakeTimeProvider();
131+
var subject = Subject.Create<int>();
132+
133+
var results = new List<int>();
134+
135+
var subscription = await subject.Values.Debounce(DueTime, timeProvider).SubscribeAsync(
136+
async (x, token) => { lock (results) results.Add(x); }, CancellationToken.None);
137+
138+
await subject.OnNextAsync(1, CancellationToken.None);
139+
await subscription.DisposeAsync();
140+
141+
// Firing the timer after disposal must not emit anything.
142+
timeProvider.Advance(DueTime);
143+
await Task.Delay(50);
144+
145+
lock (results) results.ShouldBeEmpty();
146+
}
147+
}

‎src/R3Async.Tests/R3Async.Tests.csproj‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010

1111
<ItemGroup>
1212
<PackageReference Include="coverlet.collector" Version="6.0.4" />
13+
<PackageReference Include="Microsoft.Extensions.TimeProvider.Testing" Version="9.0.0" />
1314
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
1415
<PackageReference Include="Shouldly" Version="4.3.0" />
1516
<PackageReference Include="xunit" Version="2.9.3" />

‎src/R3Async/Operators/Debounce.cs‎

Lines changed: 170 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,170 @@
1+
using System;
2+
using System.Threading;
3+
using System.Threading.Tasks;
4+
using R3Async.Internals;
5+
6+
namespace R3Async;
7+
8+
public static partial class AsyncObservable
9+
{
10+
extension<T>(AsyncObservable<T> @this)
11+
{
12+
public AsyncObservable<T> Debounce(TimeSpan dueTime, TimeProvider? timeProvider = null)
13+
=> new DebounceObservable<T>(@this, dueTime, timeProvider ?? TimeProvider.System);
14+
}
15+
}
16+
17+
internal sealed class DebounceObservable<T>(AsyncObservable<T> source, TimeSpan dueTime, TimeProvider timeProvider) : AsyncObservable<T>
18+
{
19+
protected override async ValueTask<IAsyncDisposable> SubscribeAsyncCore(AsyncObserver<T> observer, CancellationToken cancellationToken)
20+
{
21+
var subscription = new DebounceSubscription(observer, dueTime, timeProvider);
22+
try
23+
{
24+
await subscription.SubscribeAsync(source, cancellationToken);
25+
}
26+
catch
27+
{
28+
await subscription.DisposeAsync();
29+
throw;
30+
}
31+
32+
return subscription;
33+
}
34+
35+
sealed class DebounceSubscription : IAsyncDisposable
36+
{
37+
readonly AsyncObserver<T> _observer;
38+
readonly TimeSpan _dueTime;
39+
readonly SingleAssignmentAsyncDisposable _sourceDisposable = new();
40+
readonly CancellationTokenSource _disposeCts = new();
41+
readonly CancellationToken _disposeCancellationToken;
42+
readonly AsyncGate _gate = new();
43+
readonly TimeProvider _timeProvider;
44+
readonly SerialAsyncDisposable _timerDisposable = new();
45+
Optional<T> _pending;
46+
long _version;
47+
bool _sourceCompleted;
48+
bool _terminated;
49+
50+
public DebounceSubscription(AsyncObserver<T> observer, TimeSpan dueTime, TimeProvider timeProvider)
51+
{
52+
_observer = observer;
53+
_dueTime = dueTime;
54+
_timeProvider = timeProvider;
55+
_disposeCancellationToken = _disposeCts.Token;
56+
}
57+
58+
public async ValueTask SubscribeAsync(AsyncObservable<T> source, CancellationToken subscriptionToken)
59+
{
60+
var subscription = await source.SubscribeAsync(new DebounceObserver(this), subscriptionToken);
61+
await _sourceDisposable.SetDisposableAsync(subscription);
62+
}
63+
64+
async ValueTask OnNextAsync(T value)
65+
{
66+
long version;
67+
using (await _gate.LockAsync())
68+
{
69+
if (_terminated || _sourceCompleted) return;
70+
_pending = new Optional<T>(value);
71+
version = unchecked(++_version);
72+
}
73+
74+
var timerSubscription = _timeProvider.CreateTimer(static state =>
75+
{
76+
var (self, scheduledVersion) = ((DebounceSubscription, long))state!;
77+
self.OnTimerFired(scheduledVersion);
78+
}, (this, version), _dueTime, Timeout.InfiniteTimeSpan);
79+
await _timerDisposable.SetDisposableAsync(timerSubscription);
80+
}
81+
82+
async void OnTimerFired(long version)
83+
{
84+
try
85+
{
86+
using (await _gate.LockAsync())
87+
{
88+
if (_terminated || _sourceCompleted || !_pending.HasValue || _disposeCancellationToken.IsCancellationRequested) return;
89+
if (version != _version) return;
90+
var value = _pending.Value!;
91+
_pending = Optional<T>.Empty;
92+
93+
await _observer.OnNextAsync(value, _disposeCancellationToken);
94+
}
95+
}
96+
catch (Exception e)
97+
{
98+
UnhandledExceptionHandler.OnUnhandledException(e);
99+
}
100+
}
101+
102+
async ValueTask OnErrorResumeAsync(Exception error, CancellationToken cancellationToken)
103+
{
104+
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(_disposeCancellationToken, cancellationToken);
105+
using (await _gate.LockAsync())
106+
{
107+
await _observer.OnErrorResumeAsync(error, linkedCts.Token);
108+
}
109+
}
110+
111+
ValueTask OnCompletedAsync(Result result) => result.IsFailure ? CompleteAsync(result) : FlushAndCompleteAsync();
112+
113+
async ValueTask FlushAndCompleteAsync()
114+
{
115+
await _timerDisposable.DisposeAsync();
116+
using (await _gate.LockAsync())
117+
{
118+
if (_terminated || _sourceCompleted || _disposeCancellationToken.IsCancellationRequested) return;
119+
_sourceCompleted = true;
120+
_terminated = true;
121+
var pending = _pending;
122+
_pending = Optional<T>.Empty;
123+
124+
if (pending.HasValue)
125+
{
126+
await _observer.OnNextAsync(pending.Value!, _disposeCancellationToken);
127+
}
128+
129+
await _observer.OnCompletedAsync(Result.Success);
130+
}
131+
132+
await _sourceDisposable.DisposeAsync();
133+
_disposeCts.Dispose();
134+
}
135+
136+
async ValueTask CompleteAsync(Result? result)
137+
{
138+
using (await _gate.LockAsync())
139+
{
140+
if (_terminated) return;
141+
_terminated = true;
142+
_pending = Optional<T>.Empty;
143+
}
144+
145+
_disposeCts.Cancel();
146+
await _timerDisposable.DisposeAsync();
147+
if (result is not null)
148+
{
149+
using (await _gate.LockAsync())
150+
{
151+
await _observer.OnCompletedAsync(result.Value);
152+
}
153+
}
154+
await _sourceDisposable.DisposeAsync();
155+
_disposeCts.Dispose();
156+
}
157+
158+
public ValueTask DisposeAsync() => CompleteAsync(null);
159+
160+
sealed class DebounceObserver(DebounceSubscription subscription) : AsyncObserver<T>
161+
{
162+
protected override ValueTask OnNextAsyncCore(T value, CancellationToken cancellationToken)
163+
=> subscription.OnNextAsync(value);
164+
protected override ValueTask OnErrorResumeAsyncCore(Exception error, CancellationToken cancellationToken)
165+
=> subscription.OnErrorResumeAsync(error, cancellationToken);
166+
protected override ValueTask OnCompletedAsyncCore(Result result)
167+
=> subscription.OnCompletedAsync(result);
168+
}
169+
}
170+
}

0 commit comments

Comments
 (0)