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