typescript - Rxjs Subject#next asynkront eller inte?

Hej, jag undrar om följande kod körs i sekvens? D.v.s. om loggningen "Worker har slutfört uppgift C" alltid sker efter loggningen "Slutförde 3 uppgift(er)"?

Lång fråga: med scan-operatorn kan jag vara säker på att uppgifterna utförs i ordning, det är jag inte orolig för. Vad jag oroar mig för är att jag vill från den sista prenumerationen göra något först efter att uppgift C har slutförts och jag är inte säker på om platsen där o.complete() placeras garanterar det. t.ex. kommer start() att köra do.("A") -> do.("B") -> do.("C") utan att vänta på att scan-kommandona ska slutföras och köra o.complete() direkt, vilket ger ett resultat:

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)

Om det kan vara fallet, hur rättar du koden så att den uppnår det jag har beskrivit?

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

Svar #1

TLDR: Subject.next är synkron.

Reaktiva strömmar är synkrona om källan är synkron, om du inte uttryckligen gör dem asynkrona eller blandar dem med asynkrona strömmar. Inget av det händer i din kod. Några exempel:

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

Vad händer i ditt exempel? Allt inne i Observable.create kommer att köras omedelbart. När du kallar worker.do("A"); så skickar this.tasks.next(task); ut ett nytt värde och tasks-strömmen körs (synkront). Samma sak händer med B och C.

När du anropar o.complete(); slutförs start(worker)-strömmen och "Worker has finished task C" skrivs ut. Sedan utförs D av tasks-strömmen.

Du kan hitta mer omfattande information om asynkront/synkront beteende i dessa artiklar:

  • Samtidighet och asynkront beteende med RxJS
  • Reaktiva strömmar och multitrådning

Relaterade artiklar