一个可作为 ReplaySubject 但仅适用于第一个订阅者的 Rx 可观察对象？

Question

构成类似于ReplaySubject 的Rx observable 的优雅方式是什么，但只发出累积序列一次 并且仅针对第一个订阅者（当该订阅者连接时）？第一次订阅后，它应该就像普通的Subject。

这适用于 .NET 项目，但我同样感谢 JavaScript/RxJS 的答案。

我在谷歌上搜索了可能的解决方案，我即将推出自己的解决方案，最终类似于 how I approached DistinctSubject。

Answer 1

我稍微修改了similar question中的实现，并将类的名称从ReplayOnceSubject更改为ReplayFirstSubscriberOnlySubject：

public class ReplayFirstSubscriberOnlySubject<T> : ISubject<T>
{
    private readonly object _locker = new object();
    private ISubject<T> _subject = new ReplaySubject<T>();

    public void OnNext(T value) { lock (_locker) _subject.OnNext(value); }
    public void OnError(Exception error) { lock (_locker) _subject.OnError(error); }
    public void OnCompleted() { lock (_locker) _subject.OnCompleted(); }

    public IDisposable Subscribe(IObserver<T> observer)
    {
        if (observer == null) throw new ArgumentNullException(nameof(observer));
        lock (_locker)
        {
            if (_subject is ReplaySubject<T> replaySubject)
            {
                var subject = new Subject<T>();
                var subscription = subject.Subscribe(observer);
                // Now replay the buffered notifications
                replaySubject.Subscribe(subject).Dispose();
                replaySubject.Dispose();
                _subject = subject;
                return subscription;
            }
            else
                return _subject.Subscribe(observer);
        }
    }
}

这可能不是最有效的解决方案，因为每次操作都会获取两个不同的 locks（_locker 和 internal _gate），但它 也应该不会很糟糕。