queueScheduler in rx.js 6.3 is synchronous - why this example doesn't causes SO If I use queueScheduler?

user_6528560 0
Viewed 730

I have interesting example, not a real-life task but anyway:

const signal = new Subject();
let count = 0;

const somecalculations = (count) => console.log('do some calculations with ', count);


console.log('Start');
signal.pipe(take(1500)/*, observeOn(queueScheduler)*/)
  .subscribe(() => {
  somecalculations(count);
  signal.next(count++);
  console.log('check if reached ', count)
});

signal.next(count++);
console.log('Stop');

codepen

Subject.next works in synchronous way, so if i comment out observeOn(queueScheduler) - it causes Stack overflow (I control number of iterations with take operator, and on my computer if number is bigger then 1370 - it causes SO).

But if I put queueScheduler there - it works good. QueueScheduler is synchronous and somehow it allows current onNext handler run to finish running and then start next scheduled run.

Can someone explain it to me deeply with source code details? I tried to dig it but with partial success at the moment. It is about how observeOn works with QueueScheduler but answer is escaping me.

observeOn src QueueScheduler.ts asyncScheduler

rxjs
1 Answers

Thanks to cartant for support. Seems like I understood why queue scheduler is working without SO.

  1. When signal.next is called first time from observeOn _next queueScheduler.schedule->AsyncScheduler.schedule->Scheduler.schedule causes QueueAction.schedule to be called

  2. QueueAction.flush called. this.scheduler.flush - > QueueSchedulerFlush->AsyncScheduler.flush

  3. First time queue is empty and no task is executed so this.active is false. bc of this action.execute is called. Everything is called in sync way.

  4. action.execute causes onNext function to be run again. So onNext calls signal.next it goes through all 1-3 points but now this.active is true (because it is actually still previous signal.next run) and we just queue action

  5. So second signal.next is handled and we return to action.execute of first signal.next call. It works in do-while and shift actions one by one. So it finished running first signal.next action - but now we have one more in queue from second signal.next recursive call. So we run action.execute for second signal.next

  6. And situation is being repeated. First flush call manages all the other calls like: active is true, we add task to queue and then repeat to previous flush call and grab it from queue.

user_65285600
Related