Feat/debounce

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..**

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<bool> 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<int>();
+
+        var results = new List<int>();
+        var completedTcs = new TaskCompletionSource<Result>(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<int>();
+
+        var results = new List<int>();
+
+        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<int>();
+
+        var results = new List<int>();
+        var completedTcs = new TaskCompletionSource<Result>(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<int>();
+
+        var results = new List<int>();
+        var completedTcs = new TaskCompletionSource<Result>(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<int>();
+
+        var results = new List<int>();
+
+        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();
+    }
+}

src/R3Async.Tests/R3Async.Tests.csproj

@@ -10,6 +10,7 @@
 
   <ItemGroup>
     <PackageReference Include="coverlet.collector" Version="6.0.4" />
+    <PackageReference Include="Microsoft.Extensions.TimeProvider.Testing" Version="9.0.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
     <PackageReference Include="Shouldly" Version="4.3.0" />
     <PackageReference Include="xunit" Version="2.9.3" />

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<T>(AsyncObservable<T> @this)
+    {
+        public AsyncObservable<T> Debounce(TimeSpan dueTime, TimeProvider? timeProvider = null)
+            => new DebounceObservable<T>(@this, dueTime, timeProvider ?? TimeProvider.System);
+    }
+}
+
+internal sealed class DebounceObservable<T>(AsyncObservable<T> source, TimeSpan dueTime, TimeProvider timeProvider) : AsyncObservable<T>
+{
+    protected override async ValueTask<IAsyncDisposable> SubscribeAsyncCore(AsyncObserver<T> 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<T> _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<T> _pending;
+        long _version;
+        bool _sourceCompleted;
+        bool _terminated;
+
+        public DebounceSubscription(AsyncObserver<T> observer, TimeSpan dueTime, TimeProvider timeProvider)
+        {
+            _observer = observer;
+            _dueTime = dueTime;
+            _timeProvider = timeProvider;
+            _disposeCancellationToken = _disposeCts.Token;
+        }
+
+        public async ValueTask SubscribeAsync(AsyncObservable<T> 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<T>(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<T>.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<T>.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<T>.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<T>
+        {
+            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);
+        }
+    }
+}