如何创建一个 Rx observable，当最后一个观察者取消订阅时停止发布事件？

Question

我将创建一个可观察量（通过各种方式）并将其返回给感兴趣的各方，但是当他们听完后，我想拆除可观察量，这样它就不会继续消耗资源。另一种方式是将其视为在 pub 子系统中创建主题。当没有人再订阅某个主题时，您不想再保留该主题及其过滤。

Answer 1

Rx 已经有一个运算符可以满足您的需求 - 实际上是两个 - Publish & 参考计数。

以下是如何使用它们：

IObservable xs = ...

var rxs = xs.Publish().RefCount();

var sub1 = rxs.Subscribe(x => { });
var sub2 = rxs.Subscribe(x => { });

//later
sub1.Dispose();

//later 
sub2.Dispose();

//The underlying subscription to `xs` is now disposed of.

简单。

Answer 2

如果我理解你的问题，你想要创建可观察量，以便当所有订阅者都处理了他们的订阅时，即没有更多订阅者，那么你想要执行一个清理函数，这将阻止可观察量产生进一步的值。
如果这是您想要的，那么您可以执行如下操作：

//Wrap a disposable
public class WrapDisposable : IDisposable
    {
        IDisposable disp;
        Action act;
        public WrapDisposable(IDisposable _disp, Action _act)
        {
            disp = _disp;
            act = _act;
        }
        void IDisposable.Dispose()
        {
            act();
            disp.Dispose();
        }
    }

    //Observable that we want to clean up after all subs are done
    public static IObservable<long> GenerateObs(out Action cleanup)
    {
        cleanup = () =>
        {
            Console.WriteLine("All subscribers are done. Do clean up");
        };
        return Observable.Interval(TimeSpan.FromSeconds(1));
    }
    //Wrap the observable
    public static IObservable<T> WrapToClean<T>(IObservable<T> obs, Action onAllDone)
    {
        int count = 0;
        return Observable.CreateWithDisposable<T>(ob =>
        {
            var disp = obs.Subscribe(ob);
            Interlocked.Increment(ref count);
            return new WrapDisposable(disp,() =>
            {
                if (Interlocked.Decrement(ref count) == 0)
                {
                    onAllDone();                                                
                }
            });
        });
    }

//使用示例：

Action cleanup;
var obs = GenerateObs(out cleanup);
var newObs = WrapToClean(obs, cleanup);
newObs.Take(6).Subscribe(Console.WriteLine);
newObs.Take(5).Subscribe(Console.WriteLine);