如何使用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。我同意伊戈尔 - 这是一个有趣的谜题，感谢有趣的问题！