RX:在订阅启动或释放时执行操作

Question

我想跟踪IObservable的活动订阅数量吗？我想我可以通过每当"某人"调用Subscribe/Dispose时递增/递减计数器来实现这一点。

我该怎么做？或者有更好的方法吗？

这似乎是由RefCount在内部完成的，但订阅计数器没有公开。

感谢

Answer 1

最简单的方法是将Observable包装在Observable中。创建：

IObservable<string> myObs;
var returnObservable = Observable.Create<string>(subj => {
    // TODO: Write code to do stuff on Sub
    var disp = myObs.Subscribe(subj);
    return Disposable.Create(() => {
        disp.Dispose();
        // TODO: Write code to do stuff in unsub.
    });
});

Answer 2

谢谢，这正是我所需要的。这可以变成一个操作员，如下所示：

public static IObservable<TSource> OnSubscribe<TSource>(this IObservable<TSource> source, Action onSubscribe, Action onDispose)
{
    return
        Observable
            .Create<TSource>(observer =>
            {
                onSubscribe?.Invoke();
                var subscription = source.Subscribe(observer);
                return () =>
                {
                    subscription.Dispose();
                    onDispose?.Invoke();
                };
            });
}