angular - ה-RxJS observable לא מפעיל subscribe אחרי mergeMap

אני מנסה לקבץ את אוסף הנתונים שלי באמצעות אופרטורים של RxJS ולחלק אותו לזרמים רבים באפליקציית Angular שלי, אבל נראה שאני לא מצליח לגרום לזה לעבוד. ב-SignalRService שלי אני מקים טריגר של SignalR בקונסטרקטור, כך שכאשר נתונים מועברים מהשרת, הם יועברו ל-Subject שיצרתי.

export class SignalRService {
  private connection?: signalR.HubConnection;
  private orders = new Subject<OrderModel[]>();
  orders$ = this.orders.asObservable();
    
  constructor() {
    // ... SignalR connection functions ...
    
    this.connection?.on('GetOrders', (data: OrderModel[]) => {
      this.orders.next(data);
    });
  }
};

ב-OrderService אני מנוי ל-orders$ Subject באמצעות כמה אופרטורים של pipe, כי אני רוצה לפצל את הנתונים ל-3 זרמים שונים בהתאם ל-status של אובייקט ה-Order.

אני מיישר את המפה, משתמש ב-groupBy ואז ממזג שוב באמצעות המפתח והנתונים המתאימים, עם זאת, זה לא עובד מסיבה כלשהי ואני לא בטוח איפה כדאי לבדוק. כשהאני משתמש במפעילי tap בין המפעילים הנוכחיים, הוא רק רושם את שני ה-tap הראשונים. כאילו אף פעם לא מגיע ל-tap השלישי ולכן כנראה אף פעם לא מבצע את ה-subscribe. בנוסף, כש-this.orders.next(data) ב-SignalRService מתבצע פעמיים או יותר, לא קורה כלום.

export class OrderService {
  // Observable sources
  private orderCategory0 = new BehaviorSubject<OrderModel[]>([]);
  private orderCategory1 = new BehaviorSubject<OrderModel[]>([]);
  private orderCategory2 = new BehaviorSubject<OrderModel[]>([]);
  private orders = [this.orderCategory0, this.orderCategory1, this.orderCategory2];
  // Observable streams
  orderCategory0$ = this.orderCategory0.asObservable();
  orderCategory1$ = this.orderCategory1.asObservable();
  orderCategory2$ = this.orderCategory2.asObservable();
  
  constructor(private signalRService: SignalRService) {
    signalRService.orders$
      .pipe(
        mergeMap((res) => res),
        //tap((res) => console.log(res)), <-- This one shows
        groupBy((order: OrderModel) => order.status),
        //tap((res) => console.log(res)), <-- This one shows
        mergeMap((group) => zip(of(group.key), group.pipe(toArray())))
        //tap((res) => console.log(res)), <-- This one doesn't
      )
      .subscribe(([groupId, data]) => this.orders[groupId].next(data));
  }
};

שימו לב שכאשר אני עושה משהו כמו הבא ב-OrderService, הכל פועל כצפוי:

signalRService.orders$.subscribe((data: OrderModel[]) => {
  const groups = this.groupData(data);

  this.orderCategory0.next(groups[0]);
  this.orderCategory1.next(groups[1]);
  this.orderCategory2.next(groups[2]);
});

כרגע, אני אבוד, אולי אני עושה את זה לגמרי לא נכון, אז כל עצה תוערך.

ערוך: גם, כאשר אני מקודד את ההזמנות בצורה נוקשה ומשתמש ב-of(orders).pipe(...).subscribe(...), וכך מחמיץ את החלק של signalRService.order$, הכל עובד בסדר גם כן.

תשובה #1

הבעיה היא ש-group.pipe(toArray()) אף פעם לא תפיק ערכים, כי group הוא נצפה (observable) שנשאר פתוח (כנראה עד ש-order$ מושלם), ו-toArray מחכה להפיק ערכים עד שהנצפה מסתיים. עם זאת, השימוש ב-groupBy ככל הנראה היה מיותר מההתחלה.

אם אני מפרש נכון את הדוגמה שלך, אתה יוצר את הקבוצות רק כדי להעביר אותן לנושא ההתנהגות הנכון שנמצא בקולקציה הממוינת לפי מצב. אתה יכול פשוט להשתמש ב-array.reduce (או ב-groupBy של Ramda) כדי ליצור קבוצות, אין צורך להמיר את המערך ל-Observable אם אתה באמת לא צריך את הזרם.

signalRService.orders$.subscribe((orders: OrderModel[]) => {
  const groups = orders.reduce((acc, cur) => {
      (acc[cur.status] || (acc[cur.status] = [])).push(cur);
      return acc;
    },
    {} as Record<string, OrderModel[]>);
  Object.values(entries).forEach(([k, v]) => this.orders[k].next(order);
});

אם אתה באמת זקוק ל-groupBy (אולי זו הפשטה של בעיה גדולה יותר), אז אל תשתמש ב-zip ותשאף להשתמש בצינור פנימי במקום.

אתה כנראה גם לא צריך את הנושאים. אתה יכול למפות הזמנות לאובייקט הממופה ואז להציג את הקבוצות האישיות לפי הצורך.

const readonly orders$ = signalRService.orders$.pipe(
  map((orders) => orders.reduce((acc, cur) => {
      (acc[cur.status] || (acc[cur.status] = [])).push(cur);
      return acc;
    },
    {} as Record<string, OrderModel[]>)
  ),
  shareReplay(1)
);
const readonly orderCategory0$ = this.orders$.pipe(map(x => x['status0'] || []));
const readonly orderCategory1$ = this.orders$.pipe(map(x => x['status1'] || []));
const readonly orderCategory2$ = this.orders$.pipe(map(x => x['status2'] || []));