如何使用debounceTime但在某段時間後仍然會觸發該功能？

Question

TLDR; 我想使用debounceTime來執行函數只有300毫秒已經過去，沒有被調用。與此同時，我也希望能夠每1分鐘觸發一次該功能。如果過程需要很長時間。否則，該功能只會在過程結束時觸發。

---

基本上，我們的系統有一個很長的過程，會向客戶端發出大量的SignalR更新。當我在客戶端接收到服務器命令時，我會向服務器發送2個額外的HTTP請求以獲取一些信息。所以只要發送給我的服務器更新，它就會重新啓動服務器。

我使用debounceTime防止發送太多的請求回 服務器如果兩個命令之間的時間是300ms以內。但有一個 用例，其中服務器不斷向客戶端發送更新，例如1小時， 。這意味着客戶端將在1小時和 300毫秒處觸發getItemCount。

export class LeftNavigationComponent implements OnInit, OnDestroy { 
    typeACount: number = 0; 
    typeBCount: number = 0;  

    constructor(
     private itemService: service.ItemService,   
     private signalR: service.SignalRService) { } 

    ngOnInit(): void { 
     this.subscriptions = [    
      this.signalR.itemCreated.debounceTime(300).subscribe(item => this.onUpdatingData())] 
    } 

    onUpdatingData() { 
     Promise.all([ 
      this.itemService.getItemCount(APP_SETTING.TYPE_A), 
      this.itemService.getItemCount(APP_SETTING.TYPE_B)]) 
      .then(response => this.gettingCountDone(response)) 
    } 

    gettingCountDone(response) { 
     this.typeACount = <any>response[0]; 
     this.typeBCount = <any>response[1];   
    } 
} 

我還是想debounceTime防止發送太多的請求到服務器。但它應該足夠聰明，可以在接收到第一次更新後的每一分鐘內自動觸發。有沒有人有過用例？

Answer 1

以下是一張就可以了。該代碼比Pavel編寫的代碼更不優雅。

您可以在我準備的plunker中試用。 （您需要打開瀏覽器控制檯才能查看生成的輸出流）。您可能還想玩normalEventDebounceTime和forcedInterval關鍵配置參數，和/或使用sourceObservable中的事件時間。

這個想法是將merge兩個流（sourceObervable和reschedulingObservable）合成一個，這個流將被任何一個輸入觸發。每當合併的可觀察事件發出事件時，我們呼叫reschedulingSubject.next()因此推遲reschedulingObservable1000ms（因爲它的構造是debounceTime適用於Subject）。

sourceObservable被認爲是真正獨立的，即通過用戶輸入產生，或者 - 在你的情況下 - 按照我的理解，由SignalR產生。

---

const normalEventDebounceTime = 450; 
const forcedInterval = 1000; 

const sourceObservable = Rx.Observable.create(observer => { 
    setTimeout(() => observer.next('event-0'), 0); 
    setTimeout(() => observer.next('event-1'), 1000); 
    setTimeout(() => observer.next('event-2'), 1100); 
    setTimeout(() => observer.next('event-3'), 1500); 
    setTimeout(() => observer.next('event-4'), 2000); 
    setTimeout(() => observer.next('event-5'), 5000); 
    setTimeout(() => observer.next('event-6'), 8000); 
    setTimeout(() => observer.complete(), 9000); 
}); 

const reschedulingSubject = new Rx.Subject(); 

const reschedulingObservable = reschedulingSubject.asObservable().debounceTime(forcedInterval); 

const debouncedSourceObservable = sourceObservable.debounceTime(normalEventDebounceTime); 

let keepWatching = true; 

sourceObservable.subscribe(
    event => {}, 
    error => {}, 
() => { 
    keepWatching = false; 
    console.info('Source observable is complete. Stop watching please'); 
    } 
); 

Rx.Observable 
    .merge(debouncedSourceObservable, reschedulingObservable) 
    .do(() => { 
    if (keepWatching) { 
     setTimeout(() => reschedulingSubject.next('forced-next'), 100); 
    } 
    }) 
    .subscribe(event => console.info(event)); 

該代碼產生以下流：

這段代碼的

event-0 
forced-next 
event-3 
event-4 
forced-next 
forced-next 
event-5 
forced-next 
forced-next 
event-6 
Source observable is complete. Stop watching please 
forced-next 

---

優點是：

• 做幾乎正是你的問題問。 （因爲setTimeout(() => reschedulingSubject.next('forced-next'), 100)，所以說差不多）。
• 不需要自定義運算符。

缺點是：

• 的 「這樣一個簡單的問題」 相當複雜的代碼。
• 用途Subject這是IMO的最後一招。

---

同樣，你問了一個非常好的問題。總是有趣的處理這樣的難題。主演這個問題！

Answer 2

您可以使用throttleTime(60000)代替debounceTime或與debounceTime並行。要檢查這種行爲所有的球移動到開始，你會看到結果

你的情況，例如，您可以執行以下操作：

ngOnInit(): void { 
     this.subscriptions = [    
      this.signalR.itemCreated.debounceTime(300).subscribe(item => this.onUpdatingData()), 
      this.signalR.itemCreated.throttleTime(60000).subscribe(item => this.onUpdatingData()) 
     ] 
    } 

那麼方法就不會調用太頻繁，也是每分鐘一次（如果沒有事件則更少）。

也可以寫自己的實現，並結合debounceTime與throttleTime但我沒有足夠的經驗也提供了這樣的例子...

Answer 3

帕維爾的答案是接近，但如果我已經瞭解好這個問題，你想這樣：

ngOnInit(): void { 
    const debounced = this.signalR.itemCreated.debounceTime(300).share(); 
    this.subscriptions = [   
     debounced.subscribe(() => this.onUpdatingData()), 
     debounced.switchMap(() => Observable.interval(60000).takeUntil(this.signalR.itemCreated)).subscribe(() => this.onUpdatingData()) 
    ] 
} 

此代碼將做到以下幾點，當創建的項目之間的時間超過300毫秒onUpdatingData主要（）將叫做。在此之後，每當去抖動發出一個值時，1minit的throttleTime觀察值就被創建。這意味着，如果從最後一次emision開始debounced沒有發射minut，onUpdatingData（）將被執行，所以一個。

和改進將合併的觀測，因爲他們是來自同一個類型和執行相同的功能，例如像這樣：

ngOnInit(): void { 
    const debounced = this.signalR.itemCreated.debounceTime(300).share(); 
    const merged = debounced.switchMap(() => Observable.interval(60000).takeUntil(this.signalR.itemCreated)) 
    this.subscriptions = [   
     merged.subscribe(() => this.onUpdatingData()) 
    ] 
} 

我張貼出了可行的解決方案小提琴。在這個小提琴中，事件mousedown模擬流this.signalR.itemCreated。

https://jsfiddle.net/llpujol/e6b6o655/

Answer 4

這是我對了 - 如果我理解正確的問題，這我不知道的......不過，代碼如下。

// this is just simulation of source of events - in the real world it is this.signalR.itemCreated 
// values are such that they would be distinguishable from interval numbers. 
// and yes, it is Igor's idea :) 
const sourceObservable = Observable.create(observer => { 
    setTimeout(() => observer.next(100), 0); 
    setTimeout(() => observer.next(101), 1000); 
    setTimeout(() => observer.next(102), 1100); 
    setTimeout(() => observer.next(103), 1500); 
    setTimeout(() => observer.next(104), 1700); 
    setTimeout(() => observer.next(105), 2100); 
    setTimeout(() => observer.next(106), 4200); 
    setTimeout(() => observer.next(107), 5000); 
    setTimeout(() => observer.next(108), 8000); 
}); 

// debouncing too fast emits 
const itemCreated = sourceObservable.debounceTime(300); 

// starting timer after last emitted event 
// again, in the real world interval should be 1 minute, this is just for illustrative purposes 
const timeout = itemCreated.switchMap(() => Observable.interval(2000)); 

// then just merging those two 
// debounceTime(300) - to suppress possible fast timer->source consequent events 
// take(12) is just to limit example length, it is not needed in real application 
itemCreated.merge(timeout).debounceTime(300).take(12).subscribe((val) => console.log(`${val}`)); 

這將產生以下序列：

100 
// 101 skipped here by sourceObservable.debounceTime(300) 
102 
// 103 skipped here by sourceObservable.debounceTime(300) 
104 
105 
// 0 from interval() skipped here by merge().debounceTime(300) 
106 
107 
0 
108 
0 
1 
2 
3 

PS。我同意伊戈爾 - 這是一個有趣的謎題，感謝有趣的問題！