angular - RxJs: forkJoin() לא רץ כי הרשימה שלי של observables לא מלאה

ובכן, יש לי בעיה שאני לא יכול לפתור.

אני גולש ברשימת נתונים מקובץ XLSX. עבור כל שורה בקובץ נשלחות שתי בקשות לשרת:

• אחד כדי לשחזר את התקופה
• עוד כדי לשחזר משתמש

ארנק observable נוצר כדי לאחסן את שאר הנתונים בשורה. הוא הושלם מיד לאחר מכן. אחרת, הלולאה הייתה ממשיכה לפעול והנתונים האחרונים ברשימה היו נלקחים בחשבון ב-forkjoin().

אז אני משתמש ב-forkJoin() כדי לחכות לתוצאה של שלושת ה-observables שעדיין בתוך הלולאה. ברגע שהשלושת ה-observables הסתיימו, נשלח בקשה חדשה. וזה כאן שמעכב אותו. השאילתה החדשה מתווספת לרשימת ה-observables.

ואני רוצה להריץ את שאר הקוד שלי רק לאחר שרשימת ה-Observables שלי הושלמה על ידי שימוש ב-forkJoin() שוב, אחרי שעברתי על הנתונים מקובץ ה-XLSX. הבעיה היא שהלולאה שלי שעוברת על הקובץ מסתיימת לפני שרשימת ה-Observables שלי מכילה משהו, וה-forkJoin() השני אף פעם לא רץ.

קוד :

for(var objectives of this.XLSXObjectives) {
    if(objectives.values != 0) {
        // 2 requests (period and user)
        var period$ = this.storePeriods.getPeriod(new Period({month: objectives.values[0].month, year: objectives.values[0].year}));
        var user$ = this.storeUsers.getUser(new User({num_seller: objectives.values[0].userCode}));

        // Observable (wallet)
        var XLSXWalletSubject: BehaviorSubject<Wallet> = new BehaviorSubject<Wallet>(null);
        var XLSXWallet$: Observable<any> = XLSXWalletSubject.asObservable();
        XLSXWalletSubject.next(new Wallet({
            wallet_name: objectives.values[0].walletName,
            user: null,
            period: null,
            margin_m: 0,
            value_100: objectives.values[0].ValueAt100Percent,
            percentage_100_m: 0
        }));

        // Wait for the result of the three observables
        forkJoin(period$, user$, XLSXWallet$).subscribe(
            ([period, user, wallet]) => {
                console.warn("OK!");

                wallet.period = period;
                wallet.user = user;

                // New request
                var wallet$ = this.storeObjectives.addXLSXWallet(wallet);
                // Request is added to a list of observables
                this.observables$.push(wallet$);

                wallet$.subscribe(w => { ... });
            }
        );
        // Observable (wallet) complete
        XLSXWalletSubject.complete();
    }
}

console.warn(this.observables$); // this.observables$ = []
forkJoin(this.observables$).subscribe(results => {
    console.log("It doesn't work!!!")
});

אם מישהו היה יכול למצוא פתרון שיעזור לי, תודה.

תשובה #1

אני לא בטוח ב-100% כאן, אבל אני חושב שמה שאתה צריך זה משהו בכיוון הזה:

from(this.XLSXObjectives).pipe(
  filter(objectives => objectives.values != 0),
  mergeMap(objectives => forkJoin(
    this.storePeriods.getPeriod(new Period({month: objectives.values[0].month, year: objectives.values[0].year}),
    this.storeUsers.getUser(new User({num_seller: objectives.values[0].userCode}),
    of(new Wallet({
      wallet_name: objectives.values[0].walletName,
      user: null,
      period: null,
      margin_m: 0,
      value_100: objectives.values[0].ValueAt100Percent,
      percentage_100_m: 0
    }))
  )),
  map(([period, user, wallet]) => ({ ...wallet, period, user})),
  mergeMap(wallet => this.storeObjectives.addXLSXWallet(wallet)),
  toArray()
).subscribe(results => { // handle array of results here });

• המטרות נפלטות אחת אחת
• ה-mergeMap מנוי ל-forkJoin של שלושת ה-Observables הפנימיים שלך (אם אתה צריך שהבקשות יטופלו אחת אחת, השתמש ב-concatMap, ואם אתה רוצה רק מספר מסוים של בקשות במקביל, הוסף ארגומנט concurrency ל-mergeMap)
• הקריאה ל-map ממלאת את האובייקט wallet שלך עם period: period ו-user: user
• ה-mergeMap הסופי מבצע את הבקשה הבאה באמצעות האובייקט wallet
• הקריאה ל-toArray מחכה עד שה-Observable כולו יסתיים ומשדרת את כל הערכים שהתקבלו כ-arr