From 02d87fae81a421a111f9e3b934f25df7d0a2e9cd Mon Sep 17 00:00:00 2001 From: FedericoAlterio Date: Sun, 1 Mar 2026 13:01:45 +0100 Subject: [PATCH 1/2] Share + removed statelessXX variants --- README.md | 37 ++-- .../Services/InMemoryChatService.cs | 1 - src/R3Async.ConsolePlayground/Program.cs | 35 ++-- src/R3Async.Tests/Operators/MulticastTest.cs | 2 +- src/R3Async.Tests/Operators/RefCountTest.cs | 3 +- src/R3Async.Tests/Operators/ShareTest.cs | 159 ++++++++++++++++++ .../Subjects/BehaviorSubjectTest.cs | 6 +- .../Subjects/StatelessBehaviorSubjectTest.cs | 91 ---------- src/R3Async.Tests/Subjects/SubjectTest.cs | 2 +- src/R3Async/Operators/Multicast.cs | 18 -- src/R3Async/Operators/Share.cs | 151 +++++++++++++++++ src/R3Async/Subjects/BaseStatelessSubject.cs | 45 ----- .../ConcurrentStatelessReplayLatestSubject.cs | 25 --- .../Internals/ConcurrentlStatelessSubjet.cs | 24 --- .../SerialStatelessReplayLastSubject.cs | 34 ---- .../Internals/SerialStatelessSubjet.cs | 31 ---- src/R3Async/Subjects/Subject.cs | 24 +-- .../Subjects/SubjectCreationOptions.cs | 10 +- 18 files changed, 363 insertions(+), 335 deletions(-) create mode 100644 src/R3Async.Tests/Operators/ShareTest.cs delete mode 100644 src/R3Async.Tests/Subjects/StatelessBehaviorSubjectTest.cs create mode 100644 src/R3Async/Operators/Share.cs delete mode 100644 src/R3Async/Subjects/BaseStatelessSubject.cs delete mode 100644 src/R3Async/Subjects/Internals/ConcurrentStatelessReplayLatestSubject.cs delete mode 100644 src/R3Async/Subjects/Internals/ConcurrentlStatelessSubjet.cs delete mode 100644 src/R3Async/Subjects/Internals/SerialStatelessReplayLastSubject.cs delete mode 100644 src/R3Async/Subjects/Internals/SerialStatelessSubjet.cs diff --git a/README.md b/README.md index 7f6e924..02203e1 100644 --- a/README.md +++ b/README.md @@ -554,7 +554,7 @@ var multicast5 = source.Publish(initialValue: 0, new BehaviorSubjectCreationOpti The `RefCount` operator automatically manages connections to a `ConnectableAsyncObservable` based on the number of subscribers. When the first subscriber subscribes, it connects to the source. When the last subscriber unsubscribes, it disconnects. -RefCount is particularly useful with stateless subjects to create observables that automatically reset when all observers unsubscribe. +For an all-in-one operator that handles multicasting, reference counting, and automatic resetting, see the `Share` operator. ### GroupBy @@ -701,27 +701,36 @@ await producerB.DisposeAsync(); // Last reference - triggers cleanup - **Concurrent safety**: Thread-safe for concurrent access - **Cancellation support**: Factory function receives a `CancellationToken` for proper async cancellation -### Stateless Subjects +### Share -Stateless subjects are a variant of subjects that automatically reset their state when all observers unsubscribe. This is useful for creating reusable hot observables that can be "restarted" without creating a new instance. +The `Share` operator allows you to share a single subscription among multiple observers. It combines the functionality of `Publish()` and `RefCount()`, while using `ShareConfig` to configure exactly when the underlying subject should be cleared and the connection reset. -#### Stateless Subject vs Regular Subject - -- **Regular Subject**: Once completed, it stays completed. New subscribers immediately receive the completion notification. -- **Stateless Subject**: Forgets completion when all observers unsubscribe. After reset, it acts as a fresh proxy that can receive and forward new values. +```csharp +public AsyncObservable Share(ShareConfig? config = null) +public AsyncObservable Share(T startValue, ShareConfig? config = null) +public AsyncObservable Share(Func> connector, ShareConfig? config = null) +``` +`ShareConfig` has options to reset the underlying connection depending on the state of the shared subscription: -#### Stateless BehaviorSubject vs Regular BehaviorSubject +```csharp +public sealed record ShareConfig +{ + // A preconfigured config instance with all properties set to true + public static ShareConfig ResetOnCompletionAndRefCountZero { get; } -- **Regular BehaviorSubject**: Stores the latest value permanently. -- **Stateless BehaviorSubject**: Returns to its original initial value when all observers unsubscribe. + public bool ResetOnErrorResult { get; init; } + public bool ResetOnSuccessResult { get; init; } + public bool ResetOnRefCountZero { get; init; } +} +``` +By default, an empty `ShareConfig` behaves exactly like `Publish().RefCount()`: when the observable completes or the final subscriber disconnects, subsequent subscribers merely receive completion results without the observable restarting. -Stateless subjects are particularly useful with `RefCount` for creating auto-resetting multicast observables: +Using `ShareConfig` with `ResetOnRefCountZero` (or `ResetOnCompletionAndRefCountZero`) is especially useful for creating hot multicast observables that automatically restart back to their original state and re-open the source once all consumers leave. ```csharp -var source = Subject.Create(); -var refCounted = source.Values.StatelessPublish(initialValue: 0).RefCount(); +var refCounted = source.Share(startValue: 0, ShareConfig.ResetOnCompletionAndRefCountZero); // First subscription gets initial value and connects await using (await refCounted.SubscribeAsync( @@ -729,8 +738,6 @@ await using (await refCounted.SubscribeAsync( )) { // Output: First: 0 - await source.OnNextAsync(10, CancellationToken.None); - // Output: First: 10 } // All observers unsubscribed - disconnects and resets to initial value diff --git a/src/Playground/Playground.ServiceA/Services/InMemoryChatService.cs b/src/Playground/Playground.ServiceA/Services/InMemoryChatService.cs index b632a31..64769e6 100644 --- a/src/Playground/Playground.ServiceA/Services/InMemoryChatService.cs +++ b/src/Playground/Playground.ServiceA/Services/InMemoryChatService.cs @@ -14,7 +14,6 @@ public InMemoryChatService() { var chat = Subject.Create(new SubjectCreationOptions { - IsStateless = false, PublishingOption = PublishingOption.Concurrent }); diff --git a/src/R3Async.ConsolePlayground/Program.cs b/src/R3Async.ConsolePlayground/Program.cs index 60be2e1..f36b228 100644 --- a/src/R3Async.ConsolePlayground/Program.cs +++ b/src/R3Async.ConsolePlayground/Program.cs @@ -1,25 +1,18 @@ - +using R3Async; -using R3Async; -using R3Async.Subjects; -var s = Subject.Create(); -var b = from a in s.Values - group a by a into g - select g.FirstOrDefaultAsync(); -var subscription = await s.Values - .GroupBy(x => x, static x => Subject.Create()) - .Select(g => g.Take(1) - .Do(x => Console.WriteLine(x)) - .OnDispose(() => Console.WriteLine("Finished"))) - .Merge() - .SubscribeAsync(); +//var a = AsyncObservable.Create(async (observer, token) => +//{ +// await observer.OnNextAsync(1, default); +// await Task.Delay(100); +// await observer.OnCompletedAsync(Result.Failure(new InvalidOperationException())); +// return AsyncDisposable.Empty; +//}).StatelessPublish(1).RefCount(); -await s.OnNextAsync(1, default); -await s.OnNextAsync(2, default); -await s.OnNextAsync(2, default); -await s.OnNextAsync(3, default); -await s.OnNextAsync(4, default); +//var tasks = Enumerable.Range(1, 100) +// .Select(x => a.SubscribeAsync(x => Console.WriteLine(x), +// onCompleted: async r => await a.SubscribeAsync(x => Console.WriteLine($"inner {x}"), cancellationToken: default)) +// .AsTask()); -await subscription.DisposeAsync(); +//await Task.WhenAll(tasks); +//Console.ReadLine(); -Console.ReadLine(); \ No newline at end of file diff --git a/src/R3Async.Tests/Operators/MulticastTest.cs b/src/R3Async.Tests/Operators/MulticastTest.cs index 71ba69e..c4a545d 100644 --- a/src/R3Async.Tests/Operators/MulticastTest.cs +++ b/src/R3Async.Tests/Operators/MulticastTest.cs @@ -368,7 +368,7 @@ public async Task Multicast_ConcurrentSubject_Works() tcs.SetResult(); }); - var subject = Subject.Create(new SubjectCreationOptions { PublishingOption = PublishingOption.Concurrent, IsStateless = false }); + var subject = Subject.Create(new SubjectCreationOptions { PublishingOption = PublishingOption.Concurrent }); var multicast = source.Multicast(subject); var results1 = new List(); diff --git a/src/R3Async.Tests/Operators/RefCountTest.cs b/src/R3Async.Tests/Operators/RefCountTest.cs index a54a1bb..eaceea7 100644 --- a/src/R3Async.Tests/Operators/RefCountTest.cs +++ b/src/R3Async.Tests/Operators/RefCountTest.cs @@ -86,7 +86,7 @@ public async Task RefCountWithStatelessBehavior_SubscribeAfterReset_ReceivesInit { var subject = Subject.Create(); - var refCounted = subject.Values.StatelessPublish(10).RefCount(); + var refCounted = subject.Values.Share(10, ShareConfig.ResetOnCompletionAndRefCountZero); var results1 = new List(); @@ -95,7 +95,6 @@ public async Task RefCountWithStatelessBehavior_SubscribeAfterReset_ReceivesInit await subject.OnNextAsync(20, CancellationToken.None); } - // refcount is now 0, connection disposed; stateless behavior subject should reset to initial value var results2 = new List(); await using var sub2 = await refCounted.SubscribeAsync(async (x, token) => results2.Add(x), CancellationToken.None); diff --git a/src/R3Async.Tests/Operators/ShareTest.cs b/src/R3Async.Tests/Operators/ShareTest.cs new file mode 100644 index 0000000..97263ea --- /dev/null +++ b/src/R3Async.Tests/Operators/ShareTest.cs @@ -0,0 +1,159 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using R3Async.Subjects; +using Shouldly; +using Xunit; +#pragma warning disable CS1998 + +namespace R3Async.Tests.Operators; + +public class ShareTest +{ + [Fact] + public async Task Share_SharesSubscription() + { + var subscriptionCount = 0; + var source = AsyncObservable.Create(async (observer, ct) => + { + Interlocked.Increment(ref subscriptionCount); + return AsyncDisposable.Empty; + }); + + var shared = source.Share(() => Subject.Create(), new ShareConfig()); + + await using var sub1 = await shared.SubscribeAsync(); + await using var sub2 = await shared.SubscribeAsync(); + + subscriptionCount.ShouldBe(1); + } + + [Fact] + public async Task Share_ResetOnRefCountZero_ResubscribesToSource() + { + var subscriptionCount = 0; + var source = AsyncObservable.Create(async (observer, ct) => + { + Interlocked.Increment(ref subscriptionCount); + return AsyncDisposable.Empty; + }); + + var shared = source.Share(() => Subject.Create(), new ShareConfig { ResetOnRefCountZero = true }); + + await using (await shared.SubscribeAsync()) + { + subscriptionCount.ShouldBe(1); + } + + // RefCount reached zero, so next subscription should trigger resubscribe + await using (await shared.SubscribeAsync()) + { + subscriptionCount.ShouldBe(2); + } + } + + [Fact] + public async Task Share_NoResetOnRefCountZero_DoesNotResubscribeToSource() + { + var subscriptionCount = 0; + var source = AsyncObservable.Create(async (observer, ct) => + { + Interlocked.Increment(ref subscriptionCount); + return AsyncDisposable.Empty; + }); + + var shared = source.Share(() => Subject.Create(), new ShareConfig { ResetOnRefCountZero = false }); + + await using (await shared.SubscribeAsync()) + { + subscriptionCount.ShouldBe(1); + } + + // RefCount reached zero, but ResetOnRefCountZero is false + await using (await shared.SubscribeAsync()) + { + subscriptionCount.ShouldBe(1); + } + } + + [Fact] + public async Task Share_ResetOnSuccessResult_ResubscribesToSource() + { + var subscriptionCount = 0; + var source = AsyncObservable.Create(async (observer, ct) => + { + Interlocked.Increment(ref subscriptionCount); + await observer.OnCompletedAsync(Result.Success); + return AsyncDisposable.Empty; + }); + + var shared = source.Share(() => Subject.Create(), new ShareConfig { ResetOnSuccessResult = true }); + + await using var sub1 = await shared.SubscribeAsync(); + subscriptionCount.ShouldBe(1); + + await using var sub2 = await shared.SubscribeAsync(); + subscriptionCount.ShouldBe(2); + } + + [Fact] + public async Task Share_NoResetOnSuccessResult_DoesNotResubscribeToSource() + { + var subscriptionCount = 0; + var source = AsyncObservable.Create(async (observer, ct) => + { + Interlocked.Increment(ref subscriptionCount); + await observer.OnCompletedAsync(Result.Success); + return AsyncDisposable.Empty; + }); + + var shared = source.Share(() => Subject.Create(), new ShareConfig { ResetOnSuccessResult = false }); + + await using var sub1 = await shared.SubscribeAsync(); + subscriptionCount.ShouldBe(1); + + await using var sub2 = await shared.SubscribeAsync(); + subscriptionCount.ShouldBe(1); + } + + [Fact] + public async Task Share_ResetOnErrorResult_ResubscribesToSource() + { + var subscriptionCount = 0; + var source = AsyncObservable.Create(async (observer, ct) => + { + Interlocked.Increment(ref subscriptionCount); + await observer.OnCompletedAsync(Result.Failure(new Exception("test"))); + return AsyncDisposable.Empty; + }); + + var shared = source.Share(() => Subject.Create(), new ShareConfig { ResetOnErrorResult = true }); + + await using var sub1 = await shared.SubscribeAsync(); + subscriptionCount.ShouldBe(1); + + await using var sub2 = await shared.SubscribeAsync(); + subscriptionCount.ShouldBe(2); + } + + [Fact] + public async Task Share_NoResetOnErrorResult_DoesNotResubscribeToSource() + { + var subscriptionCount = 0; + var source = AsyncObservable.Create(async (observer, ct) => + { + Interlocked.Increment(ref subscriptionCount); + await observer.OnCompletedAsync(Result.Failure(new Exception("test"))); + return AsyncDisposable.Empty; + }); + + var shared = source.Share(() => Subject.Create(), new ShareConfig { ResetOnErrorResult = false }); + + await using var sub1 = await shared.SubscribeAsync(); + subscriptionCount.ShouldBe(1); + + await using var sub2 = await shared.SubscribeAsync(); + subscriptionCount.ShouldBe(1); + } +} diff --git a/src/R3Async.Tests/Subjects/BehaviorSubjectTest.cs b/src/R3Async.Tests/Subjects/BehaviorSubjectTest.cs index ab23e55..484254e 100644 --- a/src/R3Async.Tests/Subjects/BehaviorSubjectTest.cs +++ b/src/R3Async.Tests/Subjects/BehaviorSubjectTest.cs @@ -7,7 +7,7 @@ namespace R3Async.Tests.Subjects; public class BehaviorSubjectTest { static ISubject CreateBehavior(int startValue, PublishingOption option) => - Subject.CreateBehavior(startValue, new BehaviorSubjectCreationOptions { PublishingOption = option, IsStateless = false}); + Subject.CreateBehavior(startValue, new BehaviorSubjectCreationOptions { PublishingOption = option}); [Theory] [InlineData(PublishingOption.Serial)] @@ -372,7 +372,7 @@ public async Task BehaviorSubject_Reentrance_DisposeDuringInitialValue_PreventsS public async Task BehaviorSubject_ReferenceTypeStartValue_RetainsSameReference(PublishingOption option) { var startObject = new object(); - var subject = Subject.CreateBehavior(startObject, new BehaviorSubjectCreationOptions { PublishingOption = option, IsStateless = false}); + var subject = Subject.CreateBehavior(startObject, new BehaviorSubjectCreationOptions { PublishingOption = option}); object? receivedObject = null; await using var subscription = await subject.Values.SubscribeAsync( @@ -387,7 +387,7 @@ public async Task BehaviorSubject_ReferenceTypeStartValue_RetainsSameReference(P [InlineData(PublishingOption.Concurrent)] public async Task BehaviorSubject_NullStartValue_DeliversNull(PublishingOption option) { - var subject = Subject.CreateBehavior(null, new BehaviorSubjectCreationOptions { PublishingOption = option, IsStateless = false }); + var subject = Subject.CreateBehavior(null, new BehaviorSubjectCreationOptions { PublishingOption = option }); var received = false; string? receivedValue = "not null"; diff --git a/src/R3Async.Tests/Subjects/StatelessBehaviorSubjectTest.cs b/src/R3Async.Tests/Subjects/StatelessBehaviorSubjectTest.cs deleted file mode 100644 index bd195d4..0000000 --- a/src/R3Async.Tests/Subjects/StatelessBehaviorSubjectTest.cs +++ /dev/null @@ -1,91 +0,0 @@ -using R3Async.Subjects; -using Shouldly; -#pragma warning disable CS1998 - -namespace R3Async.Tests.Subjects; - -public class StatelessBehaviorSubjectTest -{ - static ISubject CreateBehavior(int startValue, PublishingOption option) => - Subject.CreateBehavior(startValue, new BehaviorSubjectCreationOptions { PublishingOption = option, IsStateless = true }); - - [Theory] - [InlineData(PublishingOption.Serial)] - [InlineData(PublishingOption.Concurrent)] - public async Task StatelessBehaviorSubject_InitialSubscriber_ReceivesStartValue(PublishingOption option) - { - var subject = CreateBehavior(42, option); - var results = new List(); - - await using var subscription = await subject.Values.SubscribeAsync( - async (x, token) => results.Add(x), - CancellationToken.None); - - results.ShouldBe(new[] { 42 }); - } - - [Theory] - [InlineData(PublishingOption.Serial)] - [InlineData(PublishingOption.Concurrent)] - public async Task StatelessBehaviorSubject_ValueResets_WhenAllSubscribersUnsubscribe(PublishingOption option) - { - var subject = CreateBehavior(1, option); - - var results1 = new List(); - var results2 = new List(); - - var sub1 = await subject.Values.SubscribeAsync( - async (x, token) => results1.Add(x), - CancellationToken.None); - - await subject.OnNextAsync(2, CancellationToken.None); - - var sub2 = await subject.Values.SubscribeAsync( - async (x, token) => results2.Add(x), - CancellationToken.None); - - await subject.OnNextAsync(3, CancellationToken.None); - - await sub1.DisposeAsync(); - await sub2.DisposeAsync(); - - var results3 = new List(); - await using var sub3 = await subject.Values.SubscribeAsync( - async (x, token) => results3.Add(x), - CancellationToken.None); - - results1.ShouldBe(new[] { 1, 2, 3 }); - results2.ShouldBe(new[] { 2, 3 }); - results3.ShouldBe(new[] { 1 }); - } - - [Theory] - [InlineData(PublishingOption.Serial)] - [InlineData(PublishingOption.Concurrent)] - public async Task StatelessBehaviorSubject_OnCompleted_ClearsSubscribersAndResetsValue(PublishingOption option) - { - var subject = CreateBehavior(1, option); - var results = new List(); - var completedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - - await using var subscription = await subject.Values.SubscribeAsync( - async (x, token) => results.Add(x), - async (ex, token) => { }, - async result => completedTcs.TrySetResult(result), - CancellationToken.None); - - await subject.OnNextAsync(2, CancellationToken.None); - await subject.OnCompletedAsync(Result.Success); - - var result = await completedTcs.Task; - result.IsSuccess.ShouldBeTrue(); - results.ShouldBe(new[] { 1, 2 }); - - var results2 = new List(); - await using var subscription2 = await subject.Values.SubscribeAsync( - async (x, token) => results2.Add(x), - CancellationToken.None); - - results2.ShouldBe(new[] { 1 }); - } -} diff --git a/src/R3Async.Tests/Subjects/SubjectTest.cs b/src/R3Async.Tests/Subjects/SubjectTest.cs index 5e909fc..f0f980e 100644 --- a/src/R3Async.Tests/Subjects/SubjectTest.cs +++ b/src/R3Async.Tests/Subjects/SubjectTest.cs @@ -6,7 +6,7 @@ namespace R3Async.Tests.Subjects; public class SubjectTest { - static ISubject Create(PublishingOption option) => Subject.Create(new SubjectCreationOptions { PublishingOption = option, IsStateless = false}); + static ISubject Create(PublishingOption option) => Subject.Create(new SubjectCreationOptions { PublishingOption = option}); [Theory] [InlineData(PublishingOption.Serial)] diff --git a/src/R3Async/Operators/Multicast.cs b/src/R3Async/Operators/Multicast.cs index 7f49ebd..47bd3f1 100644 --- a/src/R3Async/Operators/Multicast.cs +++ b/src/R3Async/Operators/Multicast.cs @@ -5,35 +5,17 @@ namespace R3Async; public static partial class AsyncObservable { - static readonly SubjectCreationOptions _statelessPublishOptions = SubjectCreationOptions.Default with - { - IsStateless = true - }; - - static readonly BehaviorSubjectCreationOptions _statelessBehaviorPublishOptions = BehaviorSubjectCreationOptions.Default with - { - IsStateless = true - }; - - static readonly ReplayLatestSubjectCreationOptions _statelessReplayLatestPublishOptions = ReplayLatestSubjectCreationOptions.Default with - { - IsStateless = true - }; - extension(AsyncObservable source) { public ConnectableAsyncObservable Multicast(ISubject subject) => new MulticastAsyncObservable(source, subject); public ConnectableAsyncObservable Publish() => source.Multicast(Subject.Create()); public ConnectableAsyncObservable Publish(SubjectCreationOptions options) => source.Multicast(Subject.Create(options)); - public ConnectableAsyncObservable StatelessPublish() => source.Multicast(Subject.Create(_statelessPublishOptions)); public ConnectableAsyncObservable Publish(T initialValue) => source.Multicast(Subject.CreateBehavior(initialValue)); public ConnectableAsyncObservable Publish(T initialValue, BehaviorSubjectCreationOptions options) => source.Multicast(Subject.CreateBehavior(initialValue, options)); - public ConnectableAsyncObservable StatelessPublish(T initialValue) => source.Multicast(Subject.CreateBehavior(initialValue, _statelessBehaviorPublishOptions)); public ConnectableAsyncObservable ReplayLatestPublish() => source.Multicast(Subject.CreateReplayLatest()); public ConnectableAsyncObservable ReplayLatestPublish(ReplayLatestSubjectCreationOptions options) => source.Multicast(Subject.CreateReplayLatest(options)); - public ConnectableAsyncObservable StatelessReplayLatestPublish() => source.Multicast(Subject.CreateReplayLatest(_statelessReplayLatestPublishOptions)); } } \ No newline at end of file diff --git a/src/R3Async/Operators/Share.cs b/src/R3Async/Operators/Share.cs new file mode 100644 index 0000000..71f5c31 --- /dev/null +++ b/src/R3Async/Operators/Share.cs @@ -0,0 +1,151 @@ +using R3Async.Internals; +using R3Async.Subjects; +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace R3Async; + +public static partial class AsyncObservable +{ + extension(AsyncObservable @this) + { + public AsyncObservable Share(ShareConfig? config = null) => @this.Share(static () => Subject.Create(), config); + public AsyncObservable Share(T startValue, ShareConfig? config = null) => @this.Share(() => Subject.CreateBehavior(startValue), config); + public AsyncObservable Share(Func> connector, ShareConfig? config = null) + { + if (@this is null) + throw new ArgumentNullException(nameof(@this)); + if (connector is null) + throw new ArgumentNullException(nameof(connector)); + + return new ShareObservable(@this, connector, config ?? new ShareConfig()); + } + } + + sealed class ShareObservable : AsyncObservable + { + readonly AsyncObservable _source; + readonly Func> _connector; + readonly ShareConfig _config; + readonly ShareSubscription _share; + + public ShareObservable(AsyncObservable source, Func> connector, ShareConfig config) + { + _source = source; + _connector = connector; + _config = config; + _share = new(this); + } + + protected override async ValueTask SubscribeAsyncCore(AsyncObserver observer, CancellationToken cancellationToken) + { + return await _share.SubscribeAsyncCore(observer, cancellationToken); + } + + sealed class ShareSubscription(ShareObservable parent) + { + readonly ShareObservable _parent = parent; + readonly AsyncGate _gate = new(); + ShareConnection? _connection; + + int _refCount; + public async ValueTask SubscribeAsyncCore(AsyncObserver observer, CancellationToken cancellationToken) + { + using (await _gate.LockAsync()) + { + _refCount++; + + var connection = _connection; + if (connection is null) + { + connection = new ShareConnection(this); + _connection = connection; + await _parent._source.SubscribeAsync(connection, cancellationToken); + } + + var disposable = await SubscribeToSubject(connection, observer, cancellationToken); + return disposable; + } + } + + async ValueTask SubscribeToSubject(ShareConnection connection, AsyncObserver observer, CancellationToken cancellationToken) + { + var subscription = await connection.Subject.Values.SubscribeAsync(observer.Wrap(), cancellationToken); + return AsyncDisposable.Create(async () => + { + var refCount = --_refCount; + await subscription.DisposeAsync(); + if (_parent._config.ResetOnRefCountZero && refCount == 0 && !connection.Completed) + { + await DisposeConnection(); + } + }); + } + + + ValueTask DisposeConnection() + { + var connection = _connection; + if (connection is null) return default; + _connection = null; + return connection.DisposeAsync(); + } + + sealed class ShareConnection(ShareSubscription parent) : AsyncObserver + { + public readonly ISubject Subject = parent._parent._connector(); + public bool Completed; + + protected override async ValueTask OnNextAsyncCore(T value, CancellationToken cancellationToken) + { + await Subject.OnNextAsync(value, cancellationToken); + } + + protected override async ValueTask OnErrorResumeAsyncCore(Exception error, CancellationToken cancellationToken) + { + await Subject.OnErrorResumeAsync(error, cancellationToken); + } + + protected override async ValueTask OnCompletedAsyncCore(Result result) + { + using (await parent._gate.LockAsync()) + { + Completed = true; + var config = parent._parent._config; + if (result.IsSuccess) + { + if (config.ResetOnSuccessResult) + { + await parent.DisposeConnection(); + } + } + else + { + if (config.ResetOnErrorResult) + { + await parent.DisposeConnection(); + } + } + + await Subject.OnCompletedAsync(result); + } + } + } + } + } +} + +public sealed record ShareConfig +{ + public static ShareConfig ResetOnCompletionAndRefCountZero { get; } = new() + { + ResetOnSuccessResult = true, + ResetOnErrorResult = true, + ResetOnRefCountZero = true + }; + + public bool ResetOnErrorResult { get; init; } + public bool ResetOnSuccessResult { get; init; } + public bool ResetOnRefCountZero { get; init; } +} diff --git a/src/R3Async/Subjects/BaseStatelessSubject.cs b/src/R3Async/Subjects/BaseStatelessSubject.cs deleted file mode 100644 index c6f6e99..0000000 --- a/src/R3Async/Subjects/BaseStatelessSubject.cs +++ /dev/null @@ -1,45 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Collections.Immutable; -using System.Threading; -using System.Threading.Tasks; - -namespace R3Async.Subjects; - -public abstract class BaseStatelessSubject : AsyncObservable, ISubject -{ - ImmutableList> _observers = []; - protected override ValueTask SubscribeAsyncCore(AsyncObserver observer, CancellationToken cancellationToken) - { - cancellationToken.ThrowIfCancellationRequested(); - var disposable = AsyncDisposable.Create(() => - { - ImmutableInterlocked.Update(ref _observers, static (observers, observer) => observers.Remove(observer), observer); - return default; - }); - - ImmutableInterlocked.Update(ref _observers, static (observers, observer) => observers.Add(observer), observer); - return new(disposable); - } - - AsyncObservable ISubject.Values => this; - public ValueTask OnNextAsync(T value, CancellationToken cancellationToken) - { - return OnNextAsyncCore(Volatile.Read(ref _observers), value, cancellationToken); - } - - protected abstract ValueTask OnNextAsyncCore(IReadOnlyList> observers, T value, CancellationToken cancellationToken); - - public ValueTask OnErrorResumeAsync(Exception error, CancellationToken cancellationToken) - { - return OnErrorResumeAsyncCore(Volatile.Read(ref _observers), error, cancellationToken); - } - - protected abstract ValueTask OnErrorResumeAsyncCore(IReadOnlyList> observers, Exception error, CancellationToken cancellationToken); - - public ValueTask OnCompletedAsync(Result result) - { - return OnCompletedAsyncCore(Volatile.Read(ref _observers), result); - } - protected abstract ValueTask OnCompletedAsyncCore(IReadOnlyList> observers, Result result); -} \ No newline at end of file diff --git a/src/R3Async/Subjects/Internals/ConcurrentStatelessReplayLatestSubject.cs b/src/R3Async/Subjects/Internals/ConcurrentStatelessReplayLatestSubject.cs deleted file mode 100644 index 49870a1..0000000 --- a/src/R3Async/Subjects/Internals/ConcurrentStatelessReplayLatestSubject.cs +++ /dev/null @@ -1,25 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; -using R3Async.Internals; - -namespace R3Async.Subjects.Internals; - -internal sealed class ConcurrentStatelessReplayLatestSubject(Optional startValue) : BaseStatelessReplayLastSubject(startValue) -{ - protected override ValueTask OnNextAsyncCore(IReadOnlyList> observers, T value, CancellationToken cancellationToken) - { - return Helpers.ForwardOnNextConcurrently(observers, value, cancellationToken); - } - - protected override ValueTask OnErrorResumeAsyncCore(IReadOnlyList> observers, Exception error, CancellationToken cancellationToken) - { - return Helpers.ForwardOnErrorResumeConcurrently(observers, error, cancellationToken); - } - - protected override ValueTask OnCompletedAsyncCore(IReadOnlyList> observers, Result result) - { - return Helpers.ForwardOnCompletedConcurrently(observers, result); - } -} diff --git a/src/R3Async/Subjects/Internals/ConcurrentlStatelessSubjet.cs b/src/R3Async/Subjects/Internals/ConcurrentlStatelessSubjet.cs deleted file mode 100644 index c97324b..0000000 --- a/src/R3Async/Subjects/Internals/ConcurrentlStatelessSubjet.cs +++ /dev/null @@ -1,24 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; - -namespace R3Async.Subjects.Internals; - -internal sealed class ConcurrentlStatelessSubjet : BaseStatelessSubject -{ - protected override ValueTask OnNextAsyncCore(IReadOnlyList> observers, T value, CancellationToken cancellationToken) - { - return Helpers.ForwardOnNextConcurrently(observers, value, cancellationToken); - } - - protected override ValueTask OnErrorResumeAsyncCore(IReadOnlyList> observers, Exception error, CancellationToken cancellationToken) - { - return Helpers.ForwardOnErrorResumeConcurrently(observers, error, cancellationToken); - } - - protected override ValueTask OnCompletedAsyncCore(IReadOnlyList> observers, Result result) - { - return Helpers.ForwardOnCompletedConcurrently(observers, result); - } -} \ No newline at end of file diff --git a/src/R3Async/Subjects/Internals/SerialStatelessReplayLastSubject.cs b/src/R3Async/Subjects/Internals/SerialStatelessReplayLastSubject.cs deleted file mode 100644 index d028e93..0000000 --- a/src/R3Async/Subjects/Internals/SerialStatelessReplayLastSubject.cs +++ /dev/null @@ -1,34 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; -using R3Async.Internals; - -namespace R3Async.Subjects.Internals; - -internal sealed class SerialStatelessReplayLastSubject(Optional startValue) : BaseStatelessReplayLastSubject(startValue) -{ - protected override async ValueTask OnNextAsyncCore(IReadOnlyList> observers, T value, CancellationToken cancellationToken) - { - foreach (var observer in observers) - { - await observer.OnNextAsync(value, cancellationToken); - } - } - - protected override async ValueTask OnErrorResumeAsyncCore(IReadOnlyList> observers, Exception error, CancellationToken cancellationToken) - { - foreach (var observer in observers) - { - await observer.OnErrorResumeAsync(error, cancellationToken); - } - } - - protected override async ValueTask OnCompletedAsyncCore(IReadOnlyList> observers, Result result) - { - foreach (var observer in observers) - { - await observer.OnCompletedAsync(result); - } - } -} diff --git a/src/R3Async/Subjects/Internals/SerialStatelessSubjet.cs b/src/R3Async/Subjects/Internals/SerialStatelessSubjet.cs deleted file mode 100644 index 964f6ae..0000000 --- a/src/R3Async/Subjects/Internals/SerialStatelessSubjet.cs +++ /dev/null @@ -1,31 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; - -namespace R3Async.Subjects.Internals; - -internal sealed class SerialStatelessSubjet : BaseStatelessSubject -{ - protected override async ValueTask OnNextAsyncCore(IReadOnlyList> observers, T value, CancellationToken cancellationToken) - { - foreach (var observer in observers) - { - await observer.OnNextAsync(value, cancellationToken); - } - } - protected override async ValueTask OnErrorResumeAsyncCore(IReadOnlyList> observers, Exception error, CancellationToken cancellationToken) - { - foreach (var observer in observers) - { - await observer.OnErrorResumeAsync(error, cancellationToken); - } - } - protected override async ValueTask OnCompletedAsyncCore(IReadOnlyList> observers, Result result) - { - foreach (var observer in observers) - { - await observer.OnCompletedAsync(result); - } - } -} \ No newline at end of file diff --git a/src/R3Async/Subjects/Subject.cs b/src/R3Async/Subjects/Subject.cs index 45b35c9..4c583f8 100644 --- a/src/R3Async/Subjects/Subject.cs +++ b/src/R3Async/Subjects/Subject.cs @@ -11,12 +11,10 @@ public static class Subject public static ISubject Create() => Create(SubjectCreationOptions.Default); public static ISubject Create(SubjectCreationOptions options) { - return (options.PublishingOption, options.IsStateless) switch + return options.PublishingOption switch { - (PublishingOption.Serial, false) => new SerialSubject(), - (PublishingOption.Concurrent, false) => new ConcurrentSubject(), - (PublishingOption.Serial, true) => new SerialStatelessSubjet(), - (PublishingOption.Concurrent, true) => new ConcurrentlStatelessSubjet(), + PublishingOption.Serial => new SerialSubject(), + PublishingOption.Concurrent => new ConcurrentSubject(), _ => throw new ArgumentOutOfRangeException() }; } @@ -24,12 +22,10 @@ public static ISubject Create(SubjectCreationOptions options) public static ISubject CreateBehavior(T startValue) => CreateBehavior(startValue, BehaviorSubjectCreationOptions.Default); public static ISubject CreateBehavior(T startValue, BehaviorSubjectCreationOptions options) { - return (options.PublishingOption, options.IsStateless) switch + return options.PublishingOption switch { - (PublishingOption.Serial, false) => new SerialReplayLatestSubject(new(startValue)), - (PublishingOption.Concurrent, false) => new ConcurrentReplayLatestSubject(new(startValue)), - (PublishingOption.Serial, true) => new SerialStatelessReplayLastSubject(new(startValue)), - (PublishingOption.Concurrent, true) => new ConcurrentStatelessReplayLatestSubject(new(startValue)), + PublishingOption.Serial => new SerialReplayLatestSubject(new(startValue)), + PublishingOption.Concurrent => new ConcurrentReplayLatestSubject(new(startValue)), _ => throw new ArgumentOutOfRangeException() }; } @@ -37,12 +33,10 @@ public static ISubject CreateBehavior(T startValue, BehaviorSubjectCreatio public static ISubject CreateReplayLatest() => CreateReplayLatest(ReplayLatestSubjectCreationOptions.Default); public static ISubject CreateReplayLatest(ReplayLatestSubjectCreationOptions options) { - return (options.PublishingOption, options.IsStateless) switch + return options.PublishingOption switch { - (PublishingOption.Serial, false) => new SerialReplayLatestSubject(Optional.Empty), - (PublishingOption.Concurrent, false) => new ConcurrentReplayLatestSubject(Optional.Empty), - (PublishingOption.Serial, true) => new SerialStatelessReplayLastSubject(Optional.Empty), - (PublishingOption.Concurrent, true) => new ConcurrentStatelessReplayLatestSubject(Optional.Empty), + PublishingOption.Serial => new SerialReplayLatestSubject(Optional.Empty), + PublishingOption.Concurrent => new ConcurrentReplayLatestSubject(Optional.Empty), _ => throw new ArgumentOutOfRangeException() }; } diff --git a/src/R3Async/Subjects/SubjectCreationOptions.cs b/src/R3Async/Subjects/SubjectCreationOptions.cs index c629cf4..b1d01fd 100644 --- a/src/R3Async/Subjects/SubjectCreationOptions.cs +++ b/src/R3Async/Subjects/SubjectCreationOptions.cs @@ -3,32 +3,26 @@ public sealed record SubjectCreationOptions { public required PublishingOption PublishingOption { get; init; } - public required bool IsStateless { get; init; } public static SubjectCreationOptions Default { get; } = new() { - PublishingOption = PublishingOption.Serial, - IsStateless = false + PublishingOption = PublishingOption.Serial }; } public sealed record BehaviorSubjectCreationOptions { public required PublishingOption PublishingOption { get; init; } - public required bool IsStateless { get; init; } public static BehaviorSubjectCreationOptions Default { get; } = new() { PublishingOption = PublishingOption.Serial, - IsStateless = false }; } public sealed record ReplayLatestSubjectCreationOptions { public required PublishingOption PublishingOption { get; init; } - public required bool IsStateless { get; init; } public static ReplayLatestSubjectCreationOptions Default { get; } = new() { - PublishingOption = PublishingOption.Serial, - IsStateless = false + PublishingOption = PublishingOption.Serial }; } \ No newline at end of file From 546fe270d88fb0b157abd9e9377b498e61a9a4cf Mon Sep 17 00:00:00 2001 From: FedericoAlterio Date: Sun, 1 Mar 2026 13:12:51 +0100 Subject: [PATCH 2/2] Added shareLatest --- src/R3Async.ConsolePlayground/Program.cs | 26 ++++++++++++------------ src/R3Async/Operators/Share.cs | 1 + 2 files changed, 14 insertions(+), 13 deletions(-) diff --git a/src/R3Async.ConsolePlayground/Program.cs b/src/R3Async.ConsolePlayground/Program.cs index f36b228..429d6a2 100644 --- a/src/R3Async.ConsolePlayground/Program.cs +++ b/src/R3Async.ConsolePlayground/Program.cs @@ -1,18 +1,18 @@ using R3Async; -//var a = AsyncObservable.Create(async (observer, token) => -//{ -// await observer.OnNextAsync(1, default); -// await Task.Delay(100); -// await observer.OnCompletedAsync(Result.Failure(new InvalidOperationException())); -// return AsyncDisposable.Empty; -//}).StatelessPublish(1).RefCount(); +var a = AsyncObservable.Create(async (observer, token) => +{ + await observer.OnNextAsync(1, default); + await Task.Delay(100); + await observer.OnCompletedAsync(Result.Failure(new InvalidOperationException("AAAA"))); + return AsyncDisposable.Empty; +}).ShareLatest(ShareConfig.ResetOnCompletionAndRefCountZero); -//var tasks = Enumerable.Range(1, 100) -// .Select(x => a.SubscribeAsync(x => Console.WriteLine(x), -// onCompleted: async r => await a.SubscribeAsync(x => Console.WriteLine($"inner {x}"), cancellationToken: default)) -// .AsTask()); +var tasks = Enumerable.Range(1, 100) + .Select(x => a.SubscribeAsync(x => Console.WriteLine(x), + onCompleted: async r => await a.SubscribeAsync(x => Console.WriteLine($"inner {x}"), cancellationToken: default)) + .AsTask()); -//await Task.WhenAll(tasks); -//Console.ReadLine(); +await Task.WhenAll(tasks); +Console.ReadLine(); diff --git a/src/R3Async/Operators/Share.cs b/src/R3Async/Operators/Share.cs index 71f5c31..f7db756 100644 --- a/src/R3Async/Operators/Share.cs +++ b/src/R3Async/Operators/Share.cs @@ -12,6 +12,7 @@ public static partial class AsyncObservable { public AsyncObservable Share(ShareConfig? config = null) => @this.Share(static () => Subject.Create(), config); public AsyncObservable Share(T startValue, ShareConfig? config = null) => @this.Share(() => Subject.CreateBehavior(startValue), config); + public AsyncObservable ShareLatest(ShareConfig? config = null) => @this.Share(Subject.CreateReplayLatest, config); public AsyncObservable Share(Func> connector, ShareConfig? config = null) { if (@this is null)