How to use debounceTime but still trigger the func

问题:

TLDR; I would like to use debounceTime to execute the function only if 300 milliseconds have passed without it being called. In the meanwhile, I also want to be able to trigger the function every 1 minutes If the process takes a long time. Otherwise, the function will only be triggered at the end of the process.

---

Basically, our system has a long process that will fire a lot of SignalR update to the client. When I received the server command on the client, I will make 2 additional HTTP requests back to server to get some information. So that It will hit the server back as long as the server sending me update.

I am using debounceTime to prevent sending too many requests back to the server If the time between two commands is within 300ms. But there is one use case where the server constantly sending the update to the client in, e.g 1 hour. Meaning the client will trigger the getItemCount at 1h and 300ms.

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

I still want to debounceTime to prevent sending too many requests to the server. But it should be smart enough to be automatically triggered every e.g 1 minutes after receiving the first updated. Has anyone had the use case before?

回答1:

Here's my take on it. The code is way way less elegant than one written by Pavel.

You can try it out in the plunker I prepared. (You'll need to open browser console to see the generated output stream). You may also want to play with the normalEventDebounceTime and forcedInterval key configuration parameters, and/or with the event timing in sourceObservable.

The idea is to merge two streams (sourceObervable and reschedulingObservable) into one which will be triggered by either of the inputs. Whenever the merged observable emits an event, we call reschedulingSubject.next() thus delaying reschedulingObservable by 1000ms (because it's constructed with debounceTime applied to Subject).

The sourceObservable is supposed to be truly independent, i.e. produced by user input, or -- in your case -- by SignalR as I understand.

---

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

That code produces the following stream:

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

---

Pros of this code are:

• Does almost exactly what you asked in the question. (Saying almost because of setTimeout(() => reschedulingSubject.next('forced-next'), 100)).
• Does not require custom operators.

Cons are:

• Pretty complex code for "such a simple problem".
• Uses Subject which is a last resort kind of a thing, IMO.

---

Again, you asked a very good question. Always interesting to deal with puzzles like this. Starring the question!

回答2:

Pavel answer is close, but If I have understood well the question, you want this:

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())
    ]
}

This code will do the following, when the time between items created are major than 300ms onUpdatingData() will be called. After that Every time debounced emits a value, a throttleTime observable of 1 minit is created. Meaning that, if debounced doesn't emit for a minut since the last emision, onUpdatingData() will be executed, and so one.

And improvement would be to merge the observables, because they are from the same type and the execute the same function, for example like this:

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())
   ]
}

I posted a fiddle showing the working solution. In this fiddle, the event mousedown simulates the stream this.signalR.itemCreated.

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

回答3:

You can use throttleTime(60000) instead of or in parallel with debounceTime. To check this behavior move all balls to the beginning and you will see the result

In your case you can for example do the following:

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

So method won't be called too often and also once per minute (or less if there is no events).

It is also possible to write your own implementation and combine debounceTime with throttleTime but I'm not experienced enough too provide such example...

回答4:

This is my take on it - if I understood the question correctly, which I'm not sure of... Nevertheless, code is below.

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

This produces the following sequence:

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. And I agree with Igor - that's an interesting brainteaser, thanks for interesting question!