rxjs5：延迟取消订阅共享可观察者

Question

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

实际观察到的太复杂，这里复制的，但对于参数的缘故：

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

我不想打了创建这个观察到昂贵的代码。我想延迟断开，直到n ms。有没有办法做到这一点？

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

当没有完成退订，再没有新的数据将被发射（除非是在流，这不是明显的在你的问题的开始触发）。 - 您的案例中的subscription1和subscription2应该收到相同的值。 如果这是设计，那么你可以不使用refCount()，但只是发布，然后做sharedObservable.connect()，在这种情况下，它总是“热”。 另一种选择可能是publishReplay(1)而不是publish()。

以任何方式，您的情况听起来有点奇怪，最有可能通过改变数据流的一般体系结构来解决 - 但是，如果不知道真实用例，很难说出哪些rxjs操作这里是最好的。

Answer 2

试图解决这个问题。下面的函数包装提供的ConnectableObservable source并维护用户的refCount。当第一用户订阅时，它呼叫connect()，然后当最后一个用户退订delayms后，从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); 
        } 
       }); 
    }) 
}