Timeout

README.md

@@ -188,6 +188,7 @@ Transform and compose observable streams:
 - `ThrottleFirst` - Emit the first value of each time window and drop the rest
 - `ThrottleLast` - Emit only the latest value of each time window, when the window expires
 - `ThrottleFirstLast` - Emit the first value of each time window immediately, then the latest value when the window expires
+- `Timeout` - Fail with a `TimeoutException` if no value is observed within a time window (measured from subscription and reset on each value)
 
 #### Concurrency & Scheduling
 - `ObserveOn` - Control execution context for downstream operators

src/R3Async.Tests/Operators/TimeoutTest.cs

@@ -0,0 +1,179 @@
+using Microsoft.Extensions.Time.Testing;
+using R3Async.Subjects;
+using Shouldly;
+#pragma warning disable CS1998
+
+namespace R3Async.Tests.Operators;
+
+public class TimeoutTest
+{
+    static readonly TimeSpan DueTime = TimeSpan.FromMilliseconds(100);
+
+    [Fact]
+    public async Task Timeout_FailsWhenNoValueArrivesInTime()
+    {
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        await using var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        timeProvider.Advance(DueTime);
+
+        var result = await completedTcs.Task;
+        result.IsFailure.ShouldBeTrue();
+        result.Exception.ShouldBeOfType<TimeoutException>();
+    }
+
+    [Fact]
+    public async Task Timeout_ValuesResetTheWindow()
+    {
+        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.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { lock (results) results.Add(x); },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        // Each value arrives before the window expires, so no timeout occurs.
+        timeProvider.Advance(TimeSpan.FromMilliseconds(50));
+        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);
+
+        completedTcs.Task.IsCompleted.ShouldBeFalse();
+        lock (results) results.ShouldBe(new[] { 1, 2, 3 });
+
+        // Now let the window expire with no value.
+        timeProvider.Advance(DueTime);
+
+        var result = await completedTcs.Task;
+        result.IsFailure.ShouldBeTrue();
+        result.Exception.ShouldBeOfType<TimeoutException>();
+    }
+
+    [Fact]
+    public async Task Timeout_ForwardsCompletionBeforeTimeout()
+    {
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        await using var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        await subject.OnCompletedAsync(Result.Success);
+
+        var result = await completedTcs.Task;
+        result.IsSuccess.ShouldBeTrue();
+
+        // A later timer expiry must not produce anything further.
+        timeProvider.Advance(DueTime);
+    }
+
+    [Fact]
+    public async Task Timeout_ForwardsSourceFailure()
+    {
+        var expected = new InvalidOperationException("boom");
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        await using var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        await subject.OnCompletedAsync(Result.Failure(expected));
+
+        var result = await completedTcs.Task;
+        result.IsFailure.ShouldBeTrue();
+        result.Exception.ShouldBe(expected);
+    }
+
+    [Fact]
+    public async Task Timeout_ForwardsOnErrorResumeWithoutTerminating()
+    {
+        var expected = new InvalidOperationException("boom");
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var results = new List<int>();
+        var errors = new List<Exception>();
+
+        await using var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { lock (results) results.Add(x); },
+            async (ex, token) => { lock (errors) errors.Add(ex); },
+            async result => { },
+            CancellationToken.None);
+
+        await subject.OnErrorResumeAsync(expected, CancellationToken.None);
+        await subject.OnNextAsync(1, CancellationToken.None);
+
+        lock (errors) errors.ShouldBe(new[] { expected });
+        lock (results) results.ShouldBe(new[] { 1 });
+    }
+
+    [Fact]
+    public async Task Timeout_DisposeCancelsTimer()
+    {
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        await subscription.DisposeAsync();
+
+        // Firing the timer after disposal must not deliver a timeout.
+        timeProvider.Advance(DueTime);
+
+        completedTcs.Task.IsCompleted.ShouldBeFalse();
+    }
+
+    [Fact]
+    public async Task Timeout_TimeoutDisposesSourceSubscription()
+    {
+        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.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { lock (results) results.Add(x); },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        timeProvider.Advance(DueTime);
+        await completedTcs.Task;
+
+        // Values pushed after the timeout must not be delivered.
+        await subject.OnNextAsync(1, CancellationToken.None);
+        lock (results) results.ShouldBeEmpty();
+    }
+}

src/R3Async.Tests/SubscribeExtensionsTest.cs

@@ -0,0 +1,52 @@
+using R3Async.Subjects;
+using Shouldly;
+#pragma warning disable CS1998
+
+namespace R3Async.Tests;
+
+public class SubscribeExtensionsTest
+{
+    [Fact]
+    public async Task SubscribeAsync_AsyncOnNextWithSyncCallbacks_ForwardsEverything()
+    {
+        var subject = Subject.Create<int>();
+
+        var results = new List<int>();
+        var errors = new List<Exception>();
+        Result? completed = null;
+
+        await using var subscription = await subject.Values.SubscribeAsync(
+            async (x, token) => results.Add(x),
+            onErrorResume: ex => errors.Add(ex),
+            onCompleted: result => completed = result,
+            CancellationToken.None);
+
+        var expectedError = new InvalidOperationException("boom");
+        await subject.OnNextAsync(1, CancellationToken.None);
+        await subject.OnErrorResumeAsync(expectedError, CancellationToken.None);
+        await subject.OnNextAsync(2, CancellationToken.None);
+        await subject.OnCompletedAsync(Result.Success);
+
+        results.ShouldBe(new[] { 1, 2 });
+        errors.ShouldBe(new[] { expectedError });
+        completed.ShouldNotBeNull();
+        completed.Value.IsSuccess.ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task SubscribeAsync_AsyncOnNextWithSyncCallbacks_NullCallbacksAreAllowed()
+    {
+        var subject = Subject.Create<int>();
+        var results = new List<int>();
+
+        await using var subscription = await subject.Values.SubscribeAsync(
+            async (x, token) => results.Add(x),
+            onErrorResume: null,
+            onCompleted: null,
+            CancellationToken.None);
+
+        await subject.OnNextAsync(42, CancellationToken.None);
+
+        results.ShouldBe(new[] { 42 });
+    }
+}

src/R3Async/AsyncObservableSubscribeExtensions.cs

@@ -62,6 +62,29 @@ public ValueTask<IAsyncDisposable> SubscribeAsync(Action<T> onNext,
             return source.SubscribeAsync(observer, cancellationToken);
         }
 
+        public ValueTask<IAsyncDisposable> SubscribeAsync(Func<T, CancellationToken, ValueTask> onNextAsync,
+                                                          Action<Exception>? onErrorResume,
+                                                          Action<Result>? onCompleted = null,
+                                                          CancellationToken cancellationToken = default)
+        {
+            if (onNextAsync is null)
+                throw new ArgumentNullException(nameof(onNextAsync));
+            if (source is null)
+                throw new ArgumentNullException(nameof(source));
+
+            var observer = new AnonymousAsyncObserver<T>(onNextAsync, onErrorResume is null ? null : (e, _) =>
+            {
+                onErrorResume(e);
+                return default;
+            }, onCompleted is null ? null : x =>
+            {
+                onCompleted(x);
+                return default;
+            });
+
+            return source.SubscribeAsync(observer, cancellationToken);
+        }
+
         public ValueTask<IAsyncDisposable> SubscribeAsync()
         {
             return source.SubscribeAsync(static (_, _)  => default, CancellationToken.None);

src/R3Async/Factories/Interval.cs

@@ -22,7 +22,7 @@ public static AsyncObservable<long> Interval(TimeSpan period, TimeProvider? time
                 else
                 {
                     var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
-                    await using var _ = timeProvider.CreateTimer(x => ((TaskCompletionSource<bool>)x!).TrySetResult(true), tcs, period, Timeout.InfiniteTimeSpan);
+                    await using var _ = timeProvider.CreateTimer(x => ((TaskCompletionSource<bool>)x!).TrySetResult(true), tcs, period, System.Threading.Timeout.InfiniteTimeSpan);
                     using var __ = cancellationToken.Register(x => ((TaskCompletionSource<bool>)x!).TrySetCanceled(cancellationToken), tcs);
                     await tcs.Task;
                 }

src/R3Async/Factories/ToAsyncObservable.cs

@@ -12,7 +12,7 @@ public static AsyncObservable<T> ToAsyncObservable<T>(this Task<T> @this)
     {
         return CreateAsBackgroundJob<T>(async (obs, cancellationToken) =>
         {
-            var result = await @this.WaitAsync(Timeout.InfiniteTimeSpan, cancellationToken);
+            var result = await @this.WaitAsync(System.Threading.Timeout.InfiniteTimeSpan, cancellationToken);
             await obs.OnNextAsync(result, cancellationToken);
             await obs.OnCompletedAsync(Result.Success);
         }, true);
@@ -22,7 +22,7 @@ public static AsyncObservable<Unit> ToAsyncObservable(this Task @this)
     {
         return CreateAsBackgroundJob<Unit>(async (obs, cancellationToken) =>
         {
-            await @this.WaitAsync(Timeout.InfiniteTimeSpan, cancellationToken);
+            await @this.WaitAsync(System.Threading.Timeout.InfiniteTimeSpan, cancellationToken);
             await obs.OnNextAsync(Unit.Default, cancellationToken);
             await obs.OnCompletedAsync(Result.Success);
         }, true);

src/R3Async/Operators/TakeUntil.cs

@@ -304,7 +304,7 @@ void Stop(Result result)
 
                     try
                     {
-                        await tcs.Task.WaitAsync(Timeout.InfiniteTimeSpan, _disposeCancellationToken);
+                        await tcs.Task.WaitAsync(System.Threading.Timeout.InfiniteTimeSpan, _disposeCancellationToken);
                         try
                         {
                             await disposable.DisposeAsync();
@@ -448,7 +448,7 @@ async void WaitAndComplete(Task task)
                 {
                     try
                     {
-                        await task.WaitAsync(Timeout.InfiniteTimeSpan, _disposeCancellationToken);
+                        await task.WaitAsync(System.Threading.Timeout.InfiniteTimeSpan, _disposeCancellationToken);
                         await ForwardOnCompletedAsync(Result.Success);
                     }
                     catch (Exception e)