Rxjs Subject#next asynchronous or not?

Question

Hi I wonder if the following code does execute in sequence? i.e. whether the logging "Worker has finished task C" does always happen after the logging "Finished 3 task(s)"?

Long question: with the scan operator, I can be sure the tasks are executed in sequence, that I am not worried. What worries me is I want from the last subscribe to do something only after task C has completed and I am not sure if the position of where o.complete() is put guarantees that. e.g. will start() runs do.("A") -> do.("B") -> do.("C") without waiting for the scans to complete and runs o.complete() straight away, giving an output:

Worker has finished task C
Doing task A
Finished 1 task(s)
Doing task B
Finished 2 task(s)
Doing task C
Finished 3 task(s)

If that can be the case, how do you fix the code so that achieves what I have described?

https://stackblitz.com/edit/typescript-xhhwme

class Worker {
  private tasks: Subject<string>;
  public init(): Observable<number> {
    this.tasks = new Subject<string>();
    return this.tasks.scan((count, task) => {
      console.log("Doing task " + task);
      return ++count;
    }, 0).asObservable();
  }
  public do(task: string): void {
    this.tasks.next(task);
  }
}

function start(worker: Worker): Observable<void> {
  return Observable.create(o => {
    const monitor = worker.init();
    monitor.subscribe(c => console.log("Finished " + c + " task(s)"));
    worker.do("A");
    worker.do("B");
    worker.do("C");
    o.complete();
    worker.do("D");
  });
}

const worker = new Worker();
start(worker).subscribe({
  complete: () => console.log("Worker has finished task C")
});

Answer 1

TLDR: Subject.next is synchronous.

Reactive streams are synchronous if the source is synchronous, unless you explicitly make them asynchronous or mix them with asynchronous streams. None of that happens in your code. Some examples:

//Synchronous
of(1,2)
  .subscribe(console.log);

//asynchronous because of async source
interval(1000)
  .subscribe(console.log);

//aynchronous because one stream is async (interval)
of(1,2)
  .pipe(
    mergeMap(x => interval(1000).pipe(take(2)))
  )
  .subscribe(console.log);

//async because we make it async
of(1,2, asyncScheduler)
  .subscribe(console.log);

What happens in your example? Everything inside Observable.create will be executed immediately. When you call worker.do("A"); then this.tasks.next(task); emits a new value and the tasks stream chain is executed (synchronously). The same happens with B and C.

When you call o.complete(); the start(worker) stream completes and "Worker has finished task C" is printed. Then D is executed by the tasks stream.

You can find more extensive information on asynchronous / synchronous behavior in these articles:

• Concurrency and Asynchronous Behavior with RxJS
• Reactive Streams and Multithreading