javascript - Det första argumentet till concat verkar inte fullbordas

Jag försöker köra concat på två observabler, men den andra observabeln i concat körs inte och jag är inte säker på varför. Det enda som händer är att done från den första observabeln skrivs ut. Jag har skapat en förenklad version av min kod på Stackblitz.

Här är början på skriptet, det kör den första observerbara i concat men inte den andra, och färdigställandemeddelandet körs aldrig:

function refresh() {
  const files = new Files();
  const deleteExisting = files.deleteExisting();
  const saveFiles = files.query().pipe(tap((i) => console.log(i)));

  concat(deleteExisting, saveFiles).subscribe({
    complete: () => console.log('Concat done!'),
  });
}

refresh();

Nästa har jag en klass som frågar databasen och strömmar filerna en i taget. Den kontrollerar om filen finns på operativsystemet; om inte läggs den till i raderingslistan (för att tas bort senare). När detta är klart skrivs done ut till konsolen, vilket förväntas.

export class Files extends DB {
  deleteExisting() {
    const toDelete: string[] = [];
    return this.stream().pipe(
      mergeMap((file) =>
        from(this.pathExists(file)).pipe(map((i) => [file, i]))
      ),
      tap<[string, boolean]>(
        ([file, exists]) => exists === false && toDelete.push(file)
      ),
      toArray(),
      map(() => toDelete.length),
      finalize(() => {
        console.log('done');
      })
    );
  }

  // The real code does `fs.pathExists()` from fs-extra which returns a promise
  pathExists(path: string) {
    return Promise.resolve(true);
  }
}

Här har vi en falsk databasklass, de faktiska uppgifterna kommer från sqlite.

• Metoden stream() returnerar resultat ett i taget (detta görs av sqlite-modulen) och jag vidarebefordrar det till subjektet.
• Metoden query() returnerar resultat som en array, och jag vidarebefordrar det till ämnet.

export class DB {
  private results1 = ['a', 'b', 'c'];
  private results2 = [1, 2, 3];

  stream() {
    const sub = new Subject<string | number>();
    // Simulate getting rows one-by-one from the db
    setTimeout(() => {
      // Runs a db call sqlite supports an `each` function which basically does this:
      this.results1.forEach((row) => sub.next(row));
      // Once each finishes a callback is called returning the number of results
      // This line is simulating that
      sub.next(this.results1.length);
    });
    return sub
      .asObservable()
      .pipe(takeWhile<string>((i) => typeof i === 'string'));
  }

  query() {
    const sub = new Subject<number[]>();
    // Simulate a callback that returns all results from database as an array
    setTimeout(() => sub.next(this.results2), 100);
    return sub.asObservable().pipe(take(1));
  }
}

Svar #1

Problemet ligger i query()-metoden.

När du prenumererar på en vanlig Subject, kommer du bara att ta emot de notifikationer som sänds efter tidpunkten för prenumerationen. Och i ditt fall sänder du bara en gång i subjectet och det är långt innan prenumerationen.

För att testa det kan du använda of-operatorn för att returnera värdena i results2

query() {
   return of(this.results2);
}

Och du kommer att se att arrayen skrivs ut.

Skål

Svar #2

Så problemet var att jag körde frågan innan jag kallade concat, vilket fick den att köras för tidigt. Att kapsla in frågan inom en Observable så att den körs när concat kallar på den istället för direkt löser problemet:

  const saveFiles = new Observable((sub) => {
    files.query().subscribe((i) => sub.next(i));
  }).pipe(tap((i) => console.log(i)));

defer verkar också fungera.

  const saveFiles = defer(() => files.query())
    .pipe(tap((i) => console.log(i)))