RxJS 按字段分组并返回新的 observable

Question

我已经关注了接口和Observable<Machine[]>，我想要实现的是按Observable<Machine[]> 中的机器symbol 属性分组并返回映射的可观察Observable<Order[]>。

export interface Machine {
    symbol: string;
    price: number;
    amount: number;
    id: number;
}

export interface Order {
    symbol: string;
    machines: OrderMachine[];
}

export interface OrderMachine {
    price: number;
    amount: number;
    id: number;
}

我尝试使用 RxJS 的 gropBy 运算符，但它似乎一个一个地返回分组数组。

machines: Machine[] = [
        { amount: 1,  id: 1, symbol: "A", price: 1 },
        { amount: 1,  id: 2, symbol: "A", price: 2 }
    ];


of(machines).pipe(
        takeUntil(this.unsubscribe),
        mergeMap(res => res),
        groupBy(m => m.symbol),
        mergeMap(group => zip(of(group.key), group.pipe(toArray()))),
        map(x => { // here I have probably wrong model [string, Machine[]]
            const orderMachines = x[1].map(y => { return <OrderMachine>{price: y.price, amount: y.amount, id: y.id }})
            return <Order>{ symbol: x[0], machines: orderMachines }  })
        );

结果我有Observable<Order>而不是Observable<Order[]>。

预期结果模型：

orders: Order[] = [
        {   
            symbol: "A", 
            machines: [
                { amount: 1, price: 1, id: 1 },
                { amount: 1, price: 2, id: 2 }
            ]
        }
    ];

Answer 1

这里有一个基于您的方法的可能解决方案，但有一些变化：

const machines = [
  { amount: 1, id: 1, symbol: "A", price: 1 },
  { amount: 1, id: 2, symbol: "A", price: 2 },
  { amount: 1, id: 3, symbol: "B", price: 3 }
];

from(machines) // (1)
  .pipe(
    // (2)
    groupBy((m) => m.symbol),
    mergeMap((group) => group.pipe(toArray())),
    map((arr) => ({
      symbol: arr[0].symbol, // every group has at least one element
      machines: arr.map(({ price, amount, id }) => ({
        price,
        amount,
        id
      }))
    })),
    toArray(), // (3)
  )
  .subscribe(console.log);

(1) 我将of(machines) 更改为from(machines)，以便将machines 中的对象一一发送到流中。在此更改之前，整个数组立即发出，因此流被破坏了。

(2) 我从管道中删除了takeUntil(this.unsubscribe) 和mergeMap(res => res)，因为没有理由在您的示例中使用它们。 takeUntil 不会产生任何影响，因为流是有限且同步的。与 mergeMap 一起应用的标识函数 (res => res) 在流中有意义，但在您的示例中并非如此。或者您的项目是否真的需要这些运算符，因为您有无限的 observable 流？

(3) toArray() 将Observable<Order> 转换为Observable<Order[]>。它一直等到流结束，并以数组的形式立即发出所有流式传输的值。

编辑：

操作员提到他宁愿需要一个与无限流兼容的解决方案，但因为toArray 仅适用于有限流，所以上面提供的答案在这种情况下永远不会发出任何东西。

为了解决这个问题，我会避免使用来自 rxjs 的groupBy。在您需要将一个流拆分为多组流的其他情况下，它是一个非常强大的工具，但在您的情况下，您只是想对一个数组进行分组，并且有更简单的方法。

this.store.pipe(
    select(fromOrder.getMachines)
    map((arr) =>
        // (*) group by symbol
        arr.reduce((acc, { symbol, price, amount, id }) => {
            acc[symbol] = {
                symbol,
                machines: (acc[symbol] ? acc[symbol].machines : [])
                    .concat({ price, amount, id })
            };
            return acc;
        }, {})
    ),
)
.subscribe((result) => 
    // (**)
    console.log(Object.values(result))
);

(*)你可以使用一个普通的groupBy implementation，它返回一个形状为{[symbol: string]: Order}的对象。

(**) result 在这里是一个对象，但您可以轻松地将其转换为数组，但应用 Object.values(result)

Answer 2

@kruschid 非常感谢您的回复，它可以正常工作，但不幸的是，当我想将它与我的商店 (ngrx) 一起使用时它不起作用，输入正常但在mergeMap 之后停止显示日志方法：

   this.store.pipe(select(fromOrder.getMachines),
                     mergeMap(res => res), // Machine[]
                     groupBy((m) => m.symbol),
                     tap(x => console.log(x)), //this shows object GroupedObservable {_isScalar: false, key: "A", groupSubject: Subject, refCountSubscription: GroupBySubscriber} 
                     mergeMap((group) => group.pipe(toArray())),
                     tap(x => console.log(x)), // this is not printed in console
                     map((arr) => <Order>({
                        symbol: arr[0].symbol,
                        machines: arr.map(({ price, amount, id }) => ({
                            price,
                            amount,
                            id
                        }))
                        })),
                     toArray())) // (3)