rxjs5：延迟取消订阅共享的 observable

Question

我有一个创建成本很高的 observable，所以我有 shared 它。但是，在某些情况下，所有订阅者都会取消订阅，然后立即（或在短暂延迟后）有新订阅者订阅。

实际的 observable 过于复杂，无法在此处复制，但为了论证：

const heavyObservable = Rx.Observable.create((observer) => {
    console.log('I am expensive, avoid hitting this code');

    return Rx.Observable
            .interval(500) // these updates are cheap though!
            .subscribe(observer)
                .add(() => {
                    console.log('Cache has been destroyed, will have to be rebuild on next call');
                });
});

我不想使用创建这个 observable 所涉及的昂贵代码。我想将断开连接延迟到 n 毫秒之后。有没有办法做到这一点？

const sharedObservable = heavyObservable
    .publish()
    // ideally I'm looking for a way to get refCount to wait for new 
    // subscribers for n ms before unsubscribing when refcount === 0
    .refCount(); 

// calling subscribe here invokes heavyObservable which can take a bit of time
const subscription1 = sharedObservable.subscribe();
// log: I am expensive, avoid hitting this code

// second call is "free" - the underlying observable is reused
const subscription2 = sharedObservable.subscribe();

subscription1.unsubscribe();
subscription2.unsubscribe();

// calling subscribe again here invokes heavyObservable over again
const subscription3 = sharedObservable.subscribe();
// log: I am expensive, avoid hitting this code

Answer 1

尝试解决这个问题。下面的函数包装了提供的 ConnectableObservable source 并维护订阅者的 refCount。它在第一个订阅者订阅时调用connect()，然后当最后一个订阅者取消订阅时，在delay ms 之后从source 调用setTimeout 和unsubscribes。

理想情况下，我宁愿修改现有的 refCount observable，但老实说我不理解代码。

不确定这是否涵盖了所有可能的边缘情况，或者它是否会产生意想不到的副作用。

Plunker：https://jsbin.com/wafahusitu/edit?js,console

function refCountWithUnsubscriptionDelay<T>(source: Rx.ConnectableObservable<T>, delay: number): Rx.Observable<T> {

    const refCount = 0;
    const sub;
    let timeoutRef;

    return Rx.Observable.create((observer: Rx.Observer<T>) => {
        refCount++;
        if (timeoutRef) {
            clearTimeout(timeoutRef);
        }
        console.log('refCount = ' + refCount);
        if (!sub) {
            // connect on first call
            sub = source.connect();
        }

        return source.subscribe(observer)
                .add(function () {
                    refCount --;
                    if (refCount <= 0) {
                        // trigger delayed unsubscription if there are no listeners
                        timeoutRef = setTimeout(() => {
                            // don't unsubscribe if new listeners have subscribed
                            if (refCount <= 0) {
                                console.log('unsub');
                                sub.unsubscribe();
                                sub = undefined;
                                timeoutRef = undefined;
                            }
                        }, delay);
                    }
                });
    })
}

Answer 2

如果没有完全取消订阅，则不会发出新数据（除非在流的开头有触发器，这在您的问题中并不明显）。 - subscription1 和 subscription2 在您的情况下应该收到相同的值。 如果这是设计使然，那么您可以根本不使用refCount()，而只是发布然后执行sharedObservable.connect()，在这种情况下它总是“热”的。 另一个选项可能是publishReplay(1) 而不是publish()。

无论如何，您的情况听起来有点奇怪，并且很可能可以通过更改数据流的一般架构来解决 - 但是在不了解实际用例的情况下，很难说出 rxjs-operation 会发生什么在这里做最好的。