Resettable Share operator like Rxjs

[fedeAlterio]

Follow-up to #2297.

There are several scenarios where this issue arises:

• An observable may fail with unknown errors (for example, one representing a SignalR, Socket.IO, or Redis Pub/Sub hub).
• Simply catching the error and retrying upstream (e.g., using Repeat) is not always acceptable. Sometimes when and how to try again should be delegated downstream (a button click for example). So it's okay terminate everything and not trying to rebuild the whole system automatically in some cases.
• The observable should be shared among downstream subscribers, but also disposed when there are no active subscribers (e.g., we don’t want to maintain a SignalR connection if no one is listening).

In my view, Rx.NET does not currently provide a built-in solution that adequately handles this type of sharing. While Multicast().RefCount() comes close, it has a key limitation: once the underlying Subject completes (especially due to an error), it cannot recover.

In contrast, RxJS provides a share operator with a configuration object that allows control over reset behavior. This includes options such as:

• resetOnError
• resetOnComplete
• resetOnRefCountZero

This approach effectively addresses the limitations of Publish().RefCount() by allowing a new subject to be created after termination, rather than attempting to reuse or reset the previous one. I think this specific operator is enough, I don't see any reason to deprecate Multicast() or what not.

Rxjs reference: https://rxjs.dev/api/operators/share

A possible implementation in .NET

using System.Reactive.Disposables;
using System.Reactive.Subjects;

namespace A

public static class ObservableEx
{
    public static IObservable<T> Share<T>(this IObservable<T> @this, ShareConfig? config = null)
    {
        if (@this is null)
            throw new ArgumentNullException(nameof(@this));

        return new ShareObservable<T>(@this, static () => new Subject<T>(), config ?? new ShareConfig());
    }

    public static IObservable<T> Share<T>(this IObservable<T> @this, Func<ISubject<T>> 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<T>(@this, connector, config ?? new ShareConfig());
    }

    sealed class ShareObservable<T> : IObservable<T>
    {
        readonly IObservable<T> _source;
        readonly Func<ISubject<T>> _connector;
        readonly ShareConfig _config;
        readonly ShareSubscription _share;

        public ShareObservable(IObservable<T> source, Func<ISubject<T>> connector, ShareConfig config)
        {
            _source = source;
            _connector = connector;
            _config = config;
            _share = new(this);
        }

        public IDisposable Subscribe(IObserver<T> observer) => _share.Subscribe(observer);

        sealed class ShareSubscription(ShareObservable<T> parent)
        {
            readonly ShareObservable<T> _parent = parent;
            readonly object _gate = new();
            ShareConnection? _connection;

            int _refCount;
            public IDisposable Subscribe(IObserver<T> observer)
            {
                lock (_gate)
                {
                    _refCount++;

                    var connection = _connection;
                    if (connection is null)
                    {
                        connection = new ShareConnection(this);
                        _connection = connection;
                        var disposable = Subscribe(connection, observer);
                        connection.Disposable.Disposable = _parent._source.SubscribeSafe(connection);
                        return disposable;
                    }
                    return Subscribe(connection, observer);
                }
            }

            IDisposable Subscribe(ShareConnection connection, IObserver<T> observer)
            {
                var subscription = connection.Subject.SubscribeSafe(observer);
                return Disposable.Create(() =>
                {
                    lock (_gate)
                    {
                        var refCount = --_refCount;
                        subscription.Dispose();
                        if (_parent._config.ResetOnRefCountZero && refCount == 0 && !connection.Completed)
                        {
                            ResetConnection();
                        }
                    }
                });
            }

            void ResetConnection()
            {
                var connection = _connection;
                if (connection is null) return;
                _connection = null;
                connection.Dispose();
            }

            sealed class ShareConnection(ShareSubscription parent) : IObserver<T>, IDisposable
            {
                public readonly ISubject<T> Subject = parent._parent._connector();
                public bool Completed;
                public readonly SingleAssignmentDisposable Disposable = new();

                public void OnCompleted()
                {
                    lock (parent._gate)
                    {
                        Completed = true;
                        var config = parent._parent._config;
                        if (config.ResetOnCompleted)
                        {
                            parent.ResetConnection();
                        }
                    }

                    Subject.OnCompleted();
                }

                public void OnError(Exception error)
                {
                    lock (parent._gate)
                    {
                        Completed = true;
                        var config = parent._parent._config;
                        if (config.ResetOnError)
                        {
                            parent.ResetConnection();
                        }
                    }

                    Subject.OnError(error);
                }

                public void OnNext(T value)
                {
                    Subject.OnNext(value);
                }

                public void Dispose() => Disposable.Dispose();
            }
        }

    }
}

public sealed record ShareConfig
{
    public static ShareConfig ResetOnCompletionAndRefCountZero { get; } = new()
    {
        ResetOnCompleted = true,
        ResetOnError = true,
        ResetOnRefCountZero = true
    };

    public bool ResetOnError { get; init; }
    public bool ResetOnCompleted { get; init; }
    public bool ResetOnRefCountZero { get; init; }
}