Share + removed statelessXX variants

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,36 +701,43 @@ 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<T> Share(ShareConfig? config = null)
+public AsyncObservable<T> Share(T startValue, ShareConfig? config = null)
+public AsyncObservable<T> Share(Func<ISubject<T>> 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<int>();
-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(
     async (value, ct) => Console.WriteLine($"First: {value}")
 ))
 {
     // Output: First: 0
-    await source.OnNextAsync(10, CancellationToken.None);
-    // Output: First: 10
 }
 // All observers unsubscribed - disconnects and resets to initial value
 

src/Playground/Playground.ServiceA/Services/InMemoryChatService.cs

@@ -14,7 +14,6 @@ public InMemoryChatService()
         {
             var chat = Subject.Create<ChatMessage>(new SubjectCreationOptions
             {
-                IsStateless = false,
                 PublishingOption = PublishingOption.Concurrent
             });
 

src/R3Async.ConsolePlayground/Program.cs

@@ -1,25 +1,18 @@
-
+using R3Async;
 
-using R3Async;
-using R3Async.Subjects;
-var s = Subject.Create<int>();
-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<int>())
- .Select(g => g.Take(1)
-               .Do(x => Console.WriteLine(x))
-               .OnDispose(() => Console.WriteLine("Finished")))
- .Merge()
- .SubscribeAsync();
+var a = AsyncObservable.Create<int>(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);
 
-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();

src/R3Async.Tests/Operators/MulticastTest.cs

@@ -368,7 +368,7 @@ public async Task Multicast_ConcurrentSubject_Works()
             tcs.SetResult();
         });
 
-        var subject = Subject.Create<int>(new SubjectCreationOptions { PublishingOption = PublishingOption.Concurrent, IsStateless = false });
+        var subject = Subject.Create<int>(new SubjectCreationOptions { PublishingOption = PublishingOption.Concurrent });
         var multicast = source.Multicast(subject);
 
         var results1 = new List<int>();

src/R3Async.Tests/Operators/RefCountTest.cs

@@ -86,7 +86,7 @@ public async Task RefCountWithStatelessBehavior_SubscribeAfterReset_ReceivesInit
     {
         var subject = Subject.Create<int>();
 
-        var refCounted = subject.Values.StatelessPublish(10).RefCount();
+        var refCounted = subject.Values.Share(10, ShareConfig.ResetOnCompletionAndRefCountZero);
 
         var results1 = new List<int>();
 
@@ -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<int>();
         await using var sub2 = await refCounted.SubscribeAsync(async (x, token) => results2.Add(x), CancellationToken.None);
 

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<int>(async (observer, ct) =>
+        {
+            Interlocked.Increment(ref subscriptionCount);
+            return AsyncDisposable.Empty;
+        });
+
+        var shared = source.Share(() => Subject.Create<int>(), 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<int>(async (observer, ct) =>
+        {
+            Interlocked.Increment(ref subscriptionCount);
+            return AsyncDisposable.Empty;
+        });
+
+        var shared = source.Share(() => Subject.Create<int>(), 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<int>(async (observer, ct) =>
+        {
+            Interlocked.Increment(ref subscriptionCount);
+            return AsyncDisposable.Empty;
+        });
+
+        var shared = source.Share(() => Subject.Create<int>(), 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<int>(async (observer, ct) =>
+        {
+            Interlocked.Increment(ref subscriptionCount);
+            await observer.OnCompletedAsync(Result.Success);
+            return AsyncDisposable.Empty;
+        });
+
+        var shared = source.Share(() => Subject.Create<int>(), 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<int>(async (observer, ct) =>
+        {
+            Interlocked.Increment(ref subscriptionCount);
+            await observer.OnCompletedAsync(Result.Success);
+            return AsyncDisposable.Empty;
+        });
+
+        var shared = source.Share(() => Subject.Create<int>(), 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<int>(async (observer, ct) =>
+        {
+            Interlocked.Increment(ref subscriptionCount);
+            await observer.OnCompletedAsync(Result.Failure(new Exception("test")));
+            return AsyncDisposable.Empty;
+        });
+
+        var shared = source.Share(() => Subject.Create<int>(), 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<int>(async (observer, ct) =>
+        {
+            Interlocked.Increment(ref subscriptionCount);
+            await observer.OnCompletedAsync(Result.Failure(new Exception("test")));
+            return AsyncDisposable.Empty;
+        });
+
+        var shared = source.Share(() => Subject.Create<int>(), new ShareConfig { ResetOnErrorResult = false });
+
+        await using var sub1 = await shared.SubscribeAsync();
+        subscriptionCount.ShouldBe(1);
+
+        await using var sub2 = await shared.SubscribeAsync();
+        subscriptionCount.ShouldBe(1);
+    }
+}

src/R3Async.Tests/Subjects/BehaviorSubjectTest.cs

@@ -7,7 +7,7 @@ namespace R3Async.Tests.Subjects;
 public class BehaviorSubjectTest
 {
     static ISubject<int> 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<string?>(null, new BehaviorSubjectCreationOptions { PublishingOption = option, IsStateless = false });
+        var subject = Subject.CreateBehavior<string?>(null, new BehaviorSubjectCreationOptions { PublishingOption = option });
         var received = false;
         string? receivedValue = "not null";