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

Question

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

Answer 1

Rx 已经有一个运算符来满足您的需求 - 实际上是两个 - Publish& RefCount。

以下是它们的使用方法：

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);