javascript - อาร์กิวเมนต์ตัวแรกของ concat ดูเหมือนจะไม่เสร็จสมบูรณ์

ฉันกำลังพยายามเรียกใช้ concat กับ observables สองตัว อย่างไรก็ตาม observable ตัวที่สองใน concat ไม่ทำงานและฉันไม่แน่ใจว่าทำไม สิ่งเดียวที่เกิดขึ้นคือ done จาก observable ตัวแรกถูกพิมพ์ออกมา ฉันได้สร้างเวอร์ชันที่เรียบง่ายของโค้ดของฉันบน Stackblitz แล้ว

นี่คือจุดเริ่มต้นของสคริปต์ มันรัน Observable ตัวแรกใน concat แต่ไม่รันตัวที่สอง และข้อความการทำงานเสร็จสิ้นไม่เคยถูกเรียกใช้:

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

ต่อไปผมมีคลาสที่สอบถามฐานข้อมูลและสตรีมไฟล์ทีละไฟล์ มันตรวจสอบว่าไฟล์อยู่ในระบบปฏิบัติการหรือไม่ หากไม่อยู่ มันจะเพิ่มไฟล์นั้นไปยังรายการลบ (เพื่อลบในภายหลัง) เมื่อสิ่งนี้เสร็จสิ้น done จะถูกพิมพ์ออกไปที่คอนโซลตามที่คาดไว้

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

ที่นี่เรามีคลาสฐานข้อมูลปลอม ข้อมูลจริงมาจาก sqlite

• เมธอด stream() จะส่งผลลัพธ์ทีละหนึ่ง (สิ่งนี้ทำโดยโมดูล sqlite) และผมส่งต่อสิ่งนั้นไปยัง subject
• เมธอด query() ส่งผลลัพธ์เป็นอาเรย์ และฉันจะส่งต่อผลลัพธ์นั้นไปยังซับเจ็กต์

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

คำตอบ #1

ปัญหาอยู่ที่วิธี query()

เมื่อสมัครรับ Subject แบบธรรมดา คุณจะได้รับการแจ้งเตือนที่ถูกส่งออกมา หลังจากช่วงเวลาที่สมัครเท่านั้น และในกรณีของคุณคุณเพียงแค่ส่งออกครั้งเดียวใน subject ซึ่งก่อนการสมัครมาก

ในการทดสอบ คุณสามารถใช้ตัวดำเนินการ of เพื่อส่งค่าของ results2

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

และคุณจะเห็นว่าอาเรย์ถูกพิมพ์ออกมา

โชคดี

คำตอบ #2

ดังนั้นปัญหาคือฉันกำลังรันคำสั่งค้นหาก่อนเรียกใช้ concat ทำให้มันทำงานเร็วเกินไป การห่อคำสั่งค้นหาไว้ภายใน Observable เพื่อให้มันทำงานเมื่อ concat เรียกใช้งานแทนที่จะทำงานทันที จึงแก้ปัญหาได้:

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

defer ก็ดูเหมือนว่าจะใช้ได้ด้วยเช่นกัน

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