angular - el observable de RxJS no activa subscribe después de mergeMap

Estoy intentando agrupar mi colección de datos usando los operadores de RxJS y dividirla en múltiples flujos en mi aplicación Angular, pero no parece funcionar. En mi SignalRService estoy configurando un disparador de SignalR en el constructor, de manera que cuando se envían datos desde el servidor, estos se pasen al Subject que creé.

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

En el OrderService me suscribo al orders$ Subject usando algunos operadores de pipe, porque quiero dividir los datos en 3 flujos diferentes basados en el status del objeto Order.

Aplano el mapa, uso groupBy y luego vuelvo a combinar usando la clave y los datos correspondientes, sin embargo, esto no funciona por alguna razón y no estoy seguro de dónde debería mirar. Cuando uso los operadores tap entre los operadores actuales, solo registra los dos primeros taps. Parece que nunca llega al tercero y, por lo tanto, supongo que nunca se ejecuta el subscribe. Además, cuando this.orders.next(data) en SignalRService se ejecuta dos veces o más, no pasa nada.

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

Tenga en cuenta que cuando hago algo como lo siguiente en OrderService, todo funciona como se espera:

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

Actualmente, estoy perdido, tal vez esté haciendo esto completamente mal, así que cualquier consejo sería apreciado.

Editar: Además, cuando codifico los Pedidos de forma fija y uso of(orders).pipe(...).subscribe(...), y por lo tanto omito la parte signalRService.order$, todo funciona bien también.

Respuesta #1

El problema es que group.pipe(toArray()) nunca va a emitir, porque group es un observable que permanece abierto (presumiblemente hasta que order$ se complete), y toArray espera para emitir hasta que el observable se complete. Dicho esto, el uso de groupBy probablemente fue excesivo desde el principio.

Si interpreto tu ejemplo correctamente, estás creando los grupos solo para pasárselos al sujeto de comportamiento correcto que está en una colección indexada por estado. Puedes simplemente usar array.reduce (o groupBy de Ramda) para crear grupos, no hay necesidad de convertir el arreglo en un observable si realmente no necesitas el flujo.

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

Si realmente necesitas el groupBy (quizás esto sea una simplificación de un problema más grande), entonces no uses zip y ten un pipe interno en su lugar.

Probablemente tampoco necesites los sujetos. Puedes asignar pedidos al objeto agrupado y luego proyectar los grupos individuales según sea necesario.

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'] || []));