diff --git a/README.md b/README.md index 02203e1..3603e3e 100644 --- a/README.md +++ b/README.md @@ -183,6 +183,9 @@ Transform and compose observable streams: - `Do` - Perform side effects - `Wrap` - Wrap observer calls +#### Timing (Throttle in classic Rx.NET) +- `Debounce` - Suppress values followed by another value within a time window; only emit after a quiet period + #### Concurrency & Scheduling - `ObserveOn` - Control execution context for downstream operators @@ -864,7 +867,7 @@ Note: `OperationCanceledException` is automatically ignored by the unhandled exc R3Async is currently under development and some features from R3 and Rx.NET are not yet implemented: -- **Throttle / Debounce** - Time-based filtering operators +- **Throttle** - **Zip** - Combine multiple observables pairwise - **Race (Amb)** - Return the first observable to emit - **Others..** diff --git a/src/R3Async.Tests/Operators/DebounceTest.cs b/src/R3Async.Tests/Operators/DebounceTest.cs new file mode 100644 index 0000000..5c9ffbb --- /dev/null +++ b/src/R3Async.Tests/Operators/DebounceTest.cs @@ -0,0 +1,147 @@ +using Microsoft.Extensions.Time.Testing; +using R3Async.Subjects; +using Shouldly; +#pragma warning disable CS1998 + +namespace R3Async.Tests.Operators; + +public class DebounceTest +{ + static readonly TimeSpan DueTime = TimeSpan.FromMilliseconds(100); + + static async Task WaitForAsync(Func condition, TimeSpan? timeout = null) + { + var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(5)); + while (!condition()) + { + if (DateTime.UtcNow > deadline) + throw new TimeoutException("Condition was not met within the timeout."); + await Task.Delay(10); + } + } + + [Fact] + public async Task Debounce_EmitsOnlyAfterQuietPeriod() + { + var timeProvider = new FakeTimeProvider(); + var subject = Subject.Create(); + + var results = new List(); + var completedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var subscription = await subject.Values.Debounce(DueTime, timeProvider).SubscribeAsync( + async (x, token) => { lock (results) results.Add(x); }, + async (ex, token) => { }, + async result => completedTcs.TrySetResult(result), + CancellationToken.None); + + // Burst of values within the due time: only the last should survive. + await subject.OnNextAsync(1, CancellationToken.None); + timeProvider.Advance(TimeSpan.FromMilliseconds(50)); + await subject.OnNextAsync(2, CancellationToken.None); + timeProvider.Advance(TimeSpan.FromMilliseconds(50)); + await subject.OnNextAsync(3, CancellationToken.None); + + // Not enough silence yet. + timeProvider.Advance(TimeSpan.FromMilliseconds(50)); + lock (results) results.ShouldBeEmpty(); + + // Now let the due time elapse without new values. + timeProvider.Advance(DueTime); + await WaitForAsync(() => { lock (results) return results.Count == 1; }); + + lock (results) results.ShouldBe(new[] { 3 }); + } + + [Fact] + public async Task Debounce_EmitsEachValueSeparatedByQuietPeriods() + { + var timeProvider = new FakeTimeProvider(); + var subject = Subject.Create(); + + var results = new List(); + + await using var subscription = await subject.Values.Debounce(DueTime, timeProvider).SubscribeAsync( + async (x, token) => { lock (results) results.Add(x); }, CancellationToken.None); + + await subject.OnNextAsync(1, CancellationToken.None); + timeProvider.Advance(DueTime); + await WaitForAsync(() => { lock (results) return results.Count == 1; }); + + await subject.OnNextAsync(2, CancellationToken.None); + timeProvider.Advance(DueTime); + await WaitForAsync(() => { lock (results) return results.Count == 2; }); + + lock (results) results.ShouldBe(new[] { 1, 2 }); + } + + [Fact] + public async Task Debounce_FlushesPendingValueOnCompletion() + { + var timeProvider = new FakeTimeProvider(); + var subject = Subject.Create(); + + var results = new List(); + var completedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var subscription = await subject.Values.Debounce(DueTime, timeProvider).SubscribeAsync( + async (x, token) => { lock (results) results.Add(x); }, + async (ex, token) => { }, + async result => completedTcs.TrySetResult(result), + CancellationToken.None); + + await subject.OnNextAsync(42, CancellationToken.None); + // Complete before the due time elapses: the pending value is flushed. + await subject.OnCompletedAsync(Result.Success); + + var result = await completedTcs.Task; + result.IsSuccess.ShouldBeTrue(); + lock (results) results.ShouldBe(new[] { 42 }); + } + + [Fact] + public async Task Debounce_ErrorDropsPendingValueAndCompletes() + { + var expected = new InvalidOperationException("boom"); + var timeProvider = new FakeTimeProvider(); + var subject = Subject.Create(); + + var results = new List(); + var completedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var subscription = await subject.Values.Debounce(DueTime, timeProvider).SubscribeAsync( + async (x, token) => { lock (results) results.Add(x); }, + async (ex, token) => { }, + async result => completedTcs.TrySetResult(result), + CancellationToken.None); + + await subject.OnNextAsync(7, CancellationToken.None); + await subject.OnCompletedAsync(Result.Failure(expected)); + + var result = await completedTcs.Task; + result.IsFailure.ShouldBeTrue(); + result.Exception.ShouldBe(expected); + lock (results) results.ShouldBeEmpty(); + } + + [Fact] + public async Task Debounce_DisposeStopsEmission() + { + var timeProvider = new FakeTimeProvider(); + var subject = Subject.Create(); + + var results = new List(); + + var subscription = await subject.Values.Debounce(DueTime, timeProvider).SubscribeAsync( + async (x, token) => { lock (results) results.Add(x); }, CancellationToken.None); + + await subject.OnNextAsync(1, CancellationToken.None); + await subscription.DisposeAsync(); + + // Firing the timer after disposal must not emit anything. + timeProvider.Advance(DueTime); + await Task.Delay(50); + + lock (results) results.ShouldBeEmpty(); + } +} diff --git a/src/R3Async.Tests/R3Async.Tests.csproj b/src/R3Async.Tests/R3Async.Tests.csproj index 2d82032..0af1327 100644 --- a/src/R3Async.Tests/R3Async.Tests.csproj +++ b/src/R3Async.Tests/R3Async.Tests.csproj @@ -10,6 +10,7 @@ + diff --git a/src/R3Async/Operators/Debounce.cs b/src/R3Async/Operators/Debounce.cs new file mode 100644 index 0000000..fd023d5 --- /dev/null +++ b/src/R3Async/Operators/Debounce.cs @@ -0,0 +1,170 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using R3Async.Internals; + +namespace R3Async; + +public static partial class AsyncObservable +{ + extension(AsyncObservable @this) + { + public AsyncObservable Debounce(TimeSpan dueTime, TimeProvider? timeProvider = null) + => new DebounceObservable(@this, dueTime, timeProvider ?? TimeProvider.System); + } +} + +internal sealed class DebounceObservable(AsyncObservable source, TimeSpan dueTime, TimeProvider timeProvider) : AsyncObservable +{ + protected override async ValueTask SubscribeAsyncCore(AsyncObserver observer, CancellationToken cancellationToken) + { + var subscription = new DebounceSubscription(observer, dueTime, timeProvider); + try + { + await subscription.SubscribeAsync(source, cancellationToken); + } + catch + { + await subscription.DisposeAsync(); + throw; + } + + return subscription; + } + + sealed class DebounceSubscription : IAsyncDisposable + { + readonly AsyncObserver _observer; + readonly TimeSpan _dueTime; + readonly SingleAssignmentAsyncDisposable _sourceDisposable = new(); + readonly CancellationTokenSource _disposeCts = new(); + readonly CancellationToken _disposeCancellationToken; + readonly AsyncGate _gate = new(); + readonly TimeProvider _timeProvider; + readonly SerialAsyncDisposable _timerDisposable = new(); + Optional _pending; + long _version; + bool _sourceCompleted; + bool _terminated; + + public DebounceSubscription(AsyncObserver observer, TimeSpan dueTime, TimeProvider timeProvider) + { + _observer = observer; + _dueTime = dueTime; + _timeProvider = timeProvider; + _disposeCancellationToken = _disposeCts.Token; + } + + public async ValueTask SubscribeAsync(AsyncObservable source, CancellationToken subscriptionToken) + { + var subscription = await source.SubscribeAsync(new DebounceObserver(this), subscriptionToken); + await _sourceDisposable.SetDisposableAsync(subscription); + } + + async ValueTask OnNextAsync(T value) + { + long version; + using (await _gate.LockAsync()) + { + if (_terminated || _sourceCompleted) return; + _pending = new Optional(value); + version = unchecked(++_version); + } + + var timerSubscription = _timeProvider.CreateTimer(static state => + { + var (self, scheduledVersion) = ((DebounceSubscription, long))state!; + self.OnTimerFired(scheduledVersion); + }, (this, version), _dueTime, Timeout.InfiniteTimeSpan); + await _timerDisposable.SetDisposableAsync(timerSubscription); + } + + async void OnTimerFired(long version) + { + try + { + using (await _gate.LockAsync()) + { + if (_terminated || _sourceCompleted || !_pending.HasValue || _disposeCancellationToken.IsCancellationRequested) return; + if (version != _version) return; + var value = _pending.Value!; + _pending = Optional.Empty; + + await _observer.OnNextAsync(value, _disposeCancellationToken); + } + } + catch (Exception e) + { + UnhandledExceptionHandler.OnUnhandledException(e); + } + } + + async ValueTask OnErrorResumeAsync(Exception error, CancellationToken cancellationToken) + { + using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(_disposeCancellationToken, cancellationToken); + using (await _gate.LockAsync()) + { + await _observer.OnErrorResumeAsync(error, linkedCts.Token); + } + } + + ValueTask OnCompletedAsync(Result result) => result.IsFailure ? CompleteAsync(result) : FlushAndCompleteAsync(); + + async ValueTask FlushAndCompleteAsync() + { + await _timerDisposable.DisposeAsync(); + using (await _gate.LockAsync()) + { + if (_terminated || _sourceCompleted || _disposeCancellationToken.IsCancellationRequested) return; + _sourceCompleted = true; + _terminated = true; + var pending = _pending; + _pending = Optional.Empty; + + if (pending.HasValue) + { + await _observer.OnNextAsync(pending.Value!, _disposeCancellationToken); + } + + await _observer.OnCompletedAsync(Result.Success); + } + + await _sourceDisposable.DisposeAsync(); + _disposeCts.Dispose(); + } + + async ValueTask CompleteAsync(Result? result) + { + using (await _gate.LockAsync()) + { + if (_terminated) return; + _terminated = true; + _pending = Optional.Empty; + } + + _disposeCts.Cancel(); + await _timerDisposable.DisposeAsync(); + if (result is not null) + { + using (await _gate.LockAsync()) + { + await _observer.OnCompletedAsync(result.Value); + } + } + await _sourceDisposable.DisposeAsync(); + _disposeCts.Dispose(); + } + + public ValueTask DisposeAsync() => CompleteAsync(null); + + sealed class DebounceObserver(DebounceSubscription subscription) : AsyncObserver + { + protected override ValueTask OnNextAsyncCore(T value, CancellationToken cancellationToken) + => subscription.OnNextAsync(value); + protected override ValueTask OnErrorResumeAsyncCore(Exception error, CancellationToken cancellationToken) + => subscription.OnErrorResumeAsync(error, cancellationToken); + protected override ValueTask OnCompletedAsyncCore(Result result) + => subscription.OnCompletedAsync(result); + } + } +}