angular - Pausa och återuppta ett observerbart flöde, vänligen föreslå bättre alternativ

Jag har ett krav på att lyssna på en ström av objekt från en observerbar. När vissa villkor uppstår kommer en asynkron uppgift att utföras på objektet och komponenten kommer att vara 'upptagen' tills detta är slutfört. Jag skulle vilja pausa hanteringen av objekt i prenumerationen tills denna uppgift har slutförts (eftersom bearbetningen av de följande objekten är beroende av resultatet) och sedan återuppta från nästa objekt i sekvensen utan någon förlust.

nästa del är förmodligen bäst att läsa medan man tittar på Plunk här

För att uppnå detta har jag använt en buffert med en switchMap. Jag trodde att dessa skulle klara jobbet på egen hand, men switchMap förstör och återskapar prenumerationen, sekvensen återställs varje gång.

export class AppComponent implements OnInit {
    source$: Observable<any>;
    clearBuffer$ = new Subject();
    busy$ = new Subject();

    private itemSubscription: Subscription;
    private stayAliveSubscription: Subscription;

    items: any[] = [];

    constructor() { }

    ngOnInit() {
      this.source$ = Observable.range(1, 500).zip(
        Observable.interval(500),
        function (x, y) { return x; }
      ).share();

      this.busy$
        .subscribe(result => {
          if (!result) {
            this.clearBuffer$.next();
          }
        }, error => {
          console.log(error);
        });
    }

    start() {
      if (!this.itemSubscription) {
        this.itemSubscription =
          this.busy$.switchMap(busy => {
            if (busy) {
              return this.source$.buffer(this.clearBuffer$);
            } else {
              return this.source$;
            }
          })
            .subscribe(items => {
              if (Array.isArray(items)) {
                this.items.push('buffered: ' + items.join());
              } else {
                this.items.push('live feed: ' + items);
              }
            }, error => {
              this.items.push(error);
            });

        this.stayAliveSubscription = this.source$
          .subscribe(result => {
            console.log(result);
          }, error => {
            console.log(error);
          });

        this.busy$.next(false);
      }
   }
...
}

För att åtgärda detta delas nu source$-observabeln och en separat prenumeration startas (stayAliveSubscription) så att en enda prenumeration används genomgående. Detta verkar rörigt för mig och jag ville fråga om någon kan visa mig bättre/alternativa tillvägagångssätt för problemet.

Jag lade arbetsprovet i en Plunk här, klicka på start för att starta prenumerationen och sedan ställa in/ta bort upptagen-växeln för att buffra och fortsätta.

redigera: fungerande kod med concatMap

Jag ändrade Plunk för att använda concatMap. Jag har klistrat in koden nedan också. Nyckeln är att den upptagna observabla som returneras i concatMap måste slutföras, du kan inte bara returnera den upptagna$observabla flera gånger och anropa next på den när statusen för upptagen ändras.

    source$: Observable<any>;
    busy$ = new Subject();
    busy: boolean;

    private itemSubscription: Subscription;
    private stayAliveSubscription: Subscription;

    items: any[] = [];

    constructor() { }

    ngOnInit() {
      this.source$ = Observable.range(1, 500).zip(
        Observable.interval(500),
        function (x, y) { return x; }
      );

      this.busy$
        .subscribe(busy => {
          this.busy = <any>busy;
        });
    }

    start() {
      if (!this.itemSubscription) {
        this.itemSubscription = this.source$.concatMap(item => {
          const busySubject = new Subject();
          this.busy$
            .subscribe(result => {
              busySubject.next(item);
              busySubject.complete();
            });

          if (this.busy) {
            return busySubject;
          } else {
            return Observable.of(item);
          }

        })
          .subscribe(item => {
            this.items.push(item);
          }, error => {
            this.items.push(error);
          });
      }

      this.setBusy(false);
    }

Svar #1

Jag förstår inte fullt ut vad du försöker göra, men om det bara handlar om att bevara ordningen på de emitterade värdena medan den "asynkrona uppgiften" kan ta lång (slumpmässig) tid, antar jag att du kan använda concatMap-operatorn.

Teori

concatMap
Projicerar varje källvärde till en Observable som slås samman i den utgående Observablen, på ett serialiserat sätt som väntar på att varje ska slutföras innan nästa slås samman.

Öva

I det här exemplet emitterar srcObservable ett värde var 100:e ms och varje värde mappas till en ny observable som emitterar ett värde mellan 0 och 2000 ms (den asynkrona uppgiften). Du kan se att ordningen är säker.

let src = Rx.Observable.timer(0,100);
src.concatMap(i=>{
  return Rx.Observable.timer(Math.random()*2000).mapTo(i); // this is the async task
}).subscribe(data=>console.log(data));

<script src="https://unpkg.com/[email /bundles/Rx.min.js"></script>

Skapa en het Observable

Du bör inte heller använda dessa prenumerationer för att få din observable att sända data. Faktiskt bör du omvandla din kalla observable till en het en med .publish() och .connect() istället för share() och subscribe() :

this.source$ = Observable.range(1, 500).zip(
        Observable.interval(500),
        function (x, y) { return x; }
      ).publish();
// blah blah blah some code
this.source$.connect();

Svar #2

delayWhen är en mycket kraftfull operator. Min lösning använder mergeMap och delayWhen.

funktioner: försök igen, reglera, pausa, återuppta

Skapa och prenumerera på Observable

const concurrentLimit = 5
const retryLimit = 10
const source$ = from(new Array(100).fill(0).map((_, i) => i))
// remove <boolean> if not typescript
const pause$ = new BehaviorSubject<boolean>(false);
const pass$ = pause$.pipe(filter((v) => !v));

const throttledTask$ = source$.pipe(
  mergeMap((item) => {
    return of(item).pipe(
      delayWhen(() => pass$),
      mergeMap(async (item) => {
         // you can also throw some errors
         return await new Promise((resolve)=>
             setTimeout(resolve(item), Math.random()*1000))
      }),
      retryWhen((errors$) => errors$.pipe(delay(1000), take(retryLimit)))
    );
  }, concurrentLimit)

const subscription = throttledTask$.subscribe(x => console.log(x))

lägg till händelsehanterare för paus/återuppta

const pause = () => { pause$.next(true) }
const resume = () => { pause$.next(false) }

Förklaring :

delayWhen kommer att pausa strömmen och vänta tills pass$-signalerna sänds.
BehaviorSubject används för att komponera pass$-signaler, vilka kommer att sända det senaste värdet när de prenumereras på.
mergeMap kan hantera asynkrona uppgifter och har en parameter för gräns för antal samtidiga trådar. När delayWhen pausar ett flöde, kommer det flödet att förbli inne i mergeMap och upptar en samtidig 'tråd'.
retryWhen kommer att prenumerera igen tills errors$.pipe(delay(1000), take(retryLimit)) emitterar klart eller fel.