I am attempting to organize my dataset in my Angular application using the RxJS operators and split it into multiple streams. However, I am facing difficulties making this work properly. Inside my SignalRService, I have set up a SignalR trigger in the constructor to pass data from the server to the Subject that I created.
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);
});
}
};
In the OrderService, I subscribe to the orders$ Subject using pipe operators to split the data into three different streams based on the status of the Order object.
I apply map flattening, groupBy, and merging based on keys and corresponding data, but for some reason, this doesn't seem to be working and I'm unsure where to troubleshoot. When I insert tap operators between the current operators, only the first two taps are displayed. It seems like the third tap is never reached, hence failing to execute the subscribe operation.
Additionally, when this.orders.next(data) is triggered more than once within SignalRService, nothing happens.
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));
}
};
If I follow a similar approach as shown in the following code snippet inside OrderService, everything works as expected:
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]);
});
Currently, I am stuck and unsure if I am approaching this problem correctly. Any guidance or suggestions would be highly appreciated.
Edit: Furthermore, when I manually input Orders and use
of(orders).pipe(...).subscribe(...)signalRService.order$ component, everything functions flawlessly.