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()を実行すると、常に「ホット」になります。 publish()ではなく、publishReplay(1)の可能性があります。

あなたのケースはちょっと変わったし、データフローの一般的なアーキテクチャの変更によって解決される可能性があります。しかし、実際の使用法を知らないと、rxjs操作ここで最高です。

Answer 2

これを解決する試み。以下の関数は、付属のConnectableObservable sourceをラップし、サブスクライバのrefCountを保持します。最初の加入者が加入したときにはconnect()を呼び出し、最後の加入者がsetTimeoutとunsubscribesをsourceからdelay msの後にコールを拒否すると、

理想的には、私は既存の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); 
        } 
       }); 
    }) 
}