Skip to content

Commit 4c3679b

Browse files
fedeAlteriofedeAlterio
authored
SubscribeXxx variants (#11)
1 parent a1eeaf2 commit 4c3679b

34 files changed

Lines changed: 1451 additions & 48 deletions

‎README.md‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -212,6 +212,29 @@ Async methods that consume the observable and return results:
212212
- `ToDictionaryAsync` - Collect to dictionary
213213
- `ToAsyncEnumerable` - Convert to async enumerable using System.Threading.Channels
214214

215+
#### Subscribe Variants
216+
217+
Aggregation operators like `FirstAsync`, `CountAsync`, `ToListAsync`, `ToAsyncEnumerable`, etc. bundle "subscribe" and "wait for the result" into a single await, so you have no way to know exactly *when* the subscription became active - only that it happened somewhere before the result arrived. That's a problem for the common **subscribe, then trigger, then wait for the response** pattern: if you send a request before you're certain the response subscription is live, the response can arrive and be missed in the gap between sending and subscribing.
218+
219+
Each of these operators has a `Subscribe`-prefixed counterpart (`SubscribeFirstAsync`, `SubscribeCountAsync`, `SubscribeToListAsync`, `SubscribeToAsyncEnumerableAsync`, ...) that splits the two steps apart: it subscribes eagerly and returns as soon as the subscription is fully established, handing back a handle whose result you await separately, whenever you're ready. This lets you subscribe first, confirm the subscription is live, *then* send whatever triggers the response, with no race window:
220+
221+
```csharp
222+
// 1. Subscribe first - by the time this await returns, we are guaranteed to observe
223+
// any matching response from this point onward.
224+
var subscription = await responses.Values.SubscribeFirstAsync(r => r.RequestId == requestId);
225+
226+
// 2. Only now send the request - the response can't arrive before we're subscribed.
227+
await SendRequestAsync(requestId);
228+
229+
// 3. Wait for the response (optionally bounded by a timeout/CancellationToken)
230+
var response = await subscription.GetResultAsync(timeout: TimeSpan.FromSeconds(5));
231+
232+
// Or give up early without ever waiting for a response
233+
await subscription.DisposeAsync();
234+
```
235+
236+
`SubscribeToAsyncEnumerableAsync` follows the same idea for the streaming case: it returns an `IAsyncDisposableReference<IAsyncEnumerable<T>>` whose `Value` is ready to enumerate immediately and whose `DisposeAsync()` unsubscribes independently of enumeration - as opposed to `ToAsyncEnumerable`, which subscribes lazily on first `MoveNextAsync`, offering no such guarantee.
237+
215238
#### ToAsyncEnumerable and Channel Selection
216239

217240
There is no "one way" to convert an async observable to an async enumerable - the behavior depends on backpressure semantics. For this reason, `ToAsyncEnumerable` accepts a channel factory function, allowing you to choose the appropriate channel type:

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

Lines changed: 0 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -145,40 +145,4 @@ public async Task Catch_Dispose_DisposesHandler()
145145

146146
handlerDisposed.ShouldBeTrue();
147147
}
148-
149-
[Fact]
150-
public async Task CatchAndIgnoreErrorResume_ForwardsOnErrorResumeToUnhandled()
151-
{
152-
var expected = new InvalidOperationException("outer resume");
153-
var tcs = new TaskCompletionSource<Exception>(TaskCreationOptions.RunContinuationsAsynchronously);
154-
155-
var source = AsyncObservable.Create<int>(async (observer, token) =>
156-
{
157-
_ = Task.Run(async () =>
158-
{
159-
await observer.OnErrorResumeAsync(expected, token);
160-
await observer.OnCompletedAsync(Result.Success);
161-
});
162-
return AsyncDisposable.Empty;
163-
});
164-
165-
var caught = source.CatchAndIgnoreErrorResume(_ => AsyncObservable.Empty<int>());
166-
167-
var field = typeof(UnhandledExceptionHandler).GetField("_unhandledException", BindingFlags.Static | BindingFlags.NonPublic);
168-
var previous = (Action<Exception>)field.GetValue(null)!;
169-
170-
try
171-
{
172-
UnhandledExceptionHandler.Register(ex => tcs.TrySetResult(ex));
173-
174-
await using var subscription = await caught.SubscribeAsync(async (x, token) => { }, CancellationToken.None);
175-
176-
var ex = await tcs.Task;
177-
ex.ShouldBe(expected);
178-
}
179-
finally
180-
{
181-
UnhandledExceptionHandler.Register(previous);
182-
}
183-
}
184148
}

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

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
using R3Async.Subjects;
2+
using Shouldly;
3+
#pragma warning disable CS1998
4+
5+
namespace R3Async.Tests.Operators;
6+
7+
public class SubscribeAnyAllAsyncTest
8+
{
9+
[Fact]
10+
public async Task SubscribeAnyAsync_MatchFound_ReturnsTrue()
11+
{
12+
var source = AsyncObservable.Create<int>((observer, token) =>
13+
{
14+
_ = Task.Run(async () =>
15+
{
16+
await observer.OnNextAsync(1, token);
17+
await observer.OnNextAsync(2, token);
18+
await observer.OnCompletedAsync(Result.Success);
19+
});
20+
return new ValueTask<IAsyncDisposable>(AsyncDisposable.Empty);
21+
});
22+
23+
await using var subscription = await source.SubscribeAnyAsync(x => x % 2 == 0);
24+
var result = await subscription.GetValueAsync();
25+
result.ShouldBeTrue();
26+
}
27+
28+
[Fact]
29+
public async Task SubscribeAnyAsync_NoMatch_ReturnsFalse()
30+
{
31+
var source = AsyncObservable.Create<int>(async (observer, token) =>
32+
{
33+
await observer.OnNextAsync(1, token);
34+
await observer.OnNextAsync(3, token);
35+
await observer.OnCompletedAsync(Result.Success);
36+
return AsyncDisposable.Empty;
37+
});
38+
39+
await using var subscription = await source.SubscribeAnyAsync(x => x % 2 == 0);
40+
var result = await subscription.GetValueAsync();
41+
result.ShouldBeFalse();
42+
}
43+
44+
[Fact]
45+
public async Task SubscribeAllAsync_AllMatch_ReturnsTrue()
46+
{
47+
var source = AsyncObservable.Create<int>(async (observer, token) =>
48+
{
49+
await observer.OnNextAsync(2, token);
50+
await observer.OnNextAsync(4, token);
51+
await observer.OnCompletedAsync(Result.Success);
52+
return AsyncDisposable.Empty;
53+
});
54+
55+
await using var subscription = await source.SubscribeAllAsync(x => x % 2 == 0);
56+
var result = await subscription.GetValueAsync();
57+
result.ShouldBeTrue();
58+
}
59+
60+
[Fact]
61+
public async Task SubscribeAllAsync_OneDoesNotMatch_ReturnsFalse()
62+
{
63+
var source = AsyncObservable.Create<int>((observer, token) =>
64+
{
65+
_ = Task.Run(async () =>
66+
{
67+
await observer.OnNextAsync(2, token);
68+
await observer.OnNextAsync(3, token);
69+
await observer.OnCompletedAsync(Result.Success);
70+
});
71+
return new ValueTask<IAsyncDisposable>(AsyncDisposable.Empty);
72+
});
73+
74+
await using var subscription = await source.SubscribeAllAsync(x => x % 2 == 0);
75+
var result = await subscription.GetValueAsync();
76+
result.ShouldBeFalse();
77+
}
78+
79+
[Fact]
80+
public async Task SubscribeAnyAsync_SubscribesEagerly_BeforeValueIsAwaited()
81+
{
82+
var subject = Subject.Create<int>();
83+
84+
var subscription = await subject.Values.SubscribeAnyAsync();
85+
86+
await subject.OnNextAsync(42, CancellationToken.None);
87+
88+
var result = await subscription.GetValueAsync();
89+
result.ShouldBeTrue();
90+
}
91+
92+
[Fact]
93+
public async Task SubscribeAnyAsync_Dispose_UnsubscribesAndFaultsValue()
94+
{
95+
var subject = Subject.Create<int>();
96+
var subscription = await subject.Values.SubscribeAnyAsync();
97+
98+
await subscription.DisposeAsync();
99+
100+
await Should.ThrowAsync<OperationCanceledException>(async () => await subscription.GetValueAsync());
101+
102+
await subject.OnNextAsync(1, CancellationToken.None);
103+
}
104+
}

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

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
using R3Async.Subjects;
2+
using Shouldly;
3+
#pragma warning disable CS1998
4+
5+
namespace R3Async.Tests.Operators;
6+
7+
public class SubscribeContainsAsyncTest
8+
{
9+
[Fact]
10+
public async Task SubscribeContainsAsync_ContainsValue_ReturnsTrue()
11+
{
12+
var source = AsyncObservable.Create<int>((observer, token) =>
13+
{
14+
_ = Task.Run(async () =>
15+
{
16+
await observer.OnNextAsync(1, token);
17+
await observer.OnNextAsync(2, token);
18+
await observer.OnNextAsync(3, token);
19+
await observer.OnCompletedAsync(Result.Success);
20+
});
21+
return new ValueTask<IAsyncDisposable>(AsyncDisposable.Empty);
22+
});
23+
24+
await using var subscription = await source.SubscribeContainsAsync(2);
25+
var result = await subscription.GetValueAsync();
26+
result.ShouldBeTrue();
27+
}
28+
29+
[Fact]
30+
public async Task SubscribeContainsAsync_DoesNotContainValue_ReturnsFalse()
31+
{
32+
var source = AsyncObservable.Create<int>(async (observer, token) =>
33+
{
34+
await observer.OnNextAsync(1, token);
35+
await observer.OnNextAsync(3, token);
36+
await observer.OnCompletedAsync(Result.Success);
37+
return AsyncDisposable.Empty;
38+
});
39+
40+
await using var subscription = await source.SubscribeContainsAsync(2);
41+
var result = await subscription.GetValueAsync();
42+
result.ShouldBeFalse();
43+
}
44+
45+
[Fact]
46+
public async Task SubscribeContainsAsync_SubscribesEagerly_BeforeValueIsAwaited()
47+
{
48+
var subject = Subject.Create<int>();
49+
50+
var subscription = await subject.Values.SubscribeContainsAsync(42);
51+
52+
await subject.OnNextAsync(42, CancellationToken.None);
53+
54+
var result = await subscription.GetValueAsync();
55+
result.ShouldBeTrue();
56+
}
57+
58+
[Fact]
59+
public async Task SubscribeContainsAsync_Dispose_UnsubscribesAndFaultsValue()
60+
{
61+
var subject = Subject.Create<int>();
62+
var subscription = await subject.Values.SubscribeContainsAsync(42);
63+
64+
await subscription.DisposeAsync();
65+
66+
await Should.ThrowAsync<OperationCanceledException>(async () => await subscription.GetValueAsync());
67+
68+
await subject.OnNextAsync(42, CancellationToken.None);
69+
}
70+
}

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

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
using R3Async.Subjects;
2+
using Shouldly;
3+
#pragma warning disable CS1998
4+
5+
namespace R3Async.Tests.Operators;
6+
7+
public class SubscribeCountAsyncTest
8+
{
9+
[Fact]
10+
public async Task SubscribeCountAsync_ReturnsMatchingCount()
11+
{
12+
var source = AsyncObservable.Create<int>((observer, token) =>
13+
{
14+
_ = Task.Run(async () =>
15+
{
16+
await observer.OnNextAsync(1, token);
17+
await observer.OnNextAsync(2, token);
18+
await observer.OnNextAsync(3, token);
19+
await observer.OnNextAsync(4, token);
20+
await observer.OnCompletedAsync(Result.Success);
21+
});
22+
return new ValueTask<IAsyncDisposable>(AsyncDisposable.Empty);
23+
});
24+
25+
await using var subscription = await source.SubscribeCountAsync(x => x % 2 == 0);
26+
var result = await subscription.GetValueAsync();
27+
result.ShouldBe(2);
28+
}
29+
30+
[Fact]
31+
public async Task SubscribeCountAsync_SubscribesEagerly_BeforeCompletionIsAwaited()
32+
{
33+
var subject = Subject.Create<int>();
34+
35+
var subscription = await subject.Values.SubscribeCountAsync();
36+
37+
await subject.OnNextAsync(1, CancellationToken.None);
38+
await subject.OnNextAsync(2, CancellationToken.None);
39+
await subject.OnCompletedAsync(Result.Success);
40+
41+
var result = await subscription.GetValueAsync();
42+
result.ShouldBe(2);
43+
}
44+
45+
[Fact]
46+
public async Task SubscribeCountAsync_Dispose_UnsubscribesAndFaultsValue()
47+
{
48+
var subject = Subject.Create<int>();
49+
var subscription = await subject.Values.SubscribeCountAsync();
50+
51+
await subject.OnNextAsync(1, CancellationToken.None);
52+
await subscription.DisposeAsync();
53+
54+
await Should.ThrowAsync<OperationCanceledException>(async () => await subscription.GetValueAsync());
55+
56+
// The underlying subscription is gone, so completion after dispose must be silently ignored.
57+
await subject.OnCompletedAsync(Result.Success);
58+
}
59+
}

0 commit comments

Comments
 (0)