问如何使用RxJs使用多个异步任务对批量操作进行管道和调优？

EN

class Product {
  id: number;
  name: string;
}

class ProductOrder {
  product: Product;
  quantity: number;
  discount: number;
  taxes: number;
}

class Order {
  productOrder: ProductOrder[];
  total: number;
}

/**
 * Check product availability, 
 * if success return the same @productOrder 
 * else throw error
 */
declare function checkProductAvailabilty(productOrder: ProductOrder): Observable<ProductOrder>;

/**
 * Checks and apply discount on products, 
 * apply changes to @productOrder 
 * if success return the same @productOrder 
 * else throw error
 */
declare function checkAndApplyDiscount(productOrder: ProductOrder): Observable<ProductOrder>;

/**
 * Apply applicable taxes on products, 
 * apply changes to @productOrder 
 * if success return the same @productOrder 
 * else throw error
 */
declare function applyApplicableTaxes(productOrder: ProductOrder): Observable<ProductOrder>;

/**
 * Calculate order total, 
 * if success return @order
 * else throw error
 */
declare function calculatOrderTotal(productOrder: ProductOrder[]): Observable<Order>;

/**
 * Create order, 
 * if success return
 * else throw error
 */
declare function createOrder(order: Order): Observable<void>;

import {
  concatMap,
  mergeMap,
  mergeAll,
  toArray,
  bufferCount,
  catchError,
} from 'rxjs/operators';


let productOrders: ProductOrder[];

from(productOrders)
  .pipe(
    concatMap(checkProductAvailabilty),
    bufferCount(10),
    mergeAll(),
    concatMap(checkAndApplyDiscount),
    bufferCount(50),
    mergeAll(),
    concatMap(applyApplicableTaxes),
    toArray(),
    mergeMap(calculatOrderTotal),
    mergeMap(createOrder),
    catchError(errorHandler)
  );

// this is the function that simulates a remote call to get the avalability of the 
// quantity requested for a certain item
function checkAvailability(prodId: string, qty: number) {
  // randomly decide if a product is available or not
  const prodAvailable = Math.random() > 0
  return  of(prodAvailable)
}

// this is the function that simulates a call to get the discound for batches of 10 products
// at the time (the last batch can have less then 10 products)
function discount(prodIds: string[]) {
  console.log('discount batch len', prodIds.length)
  // discount is fixed to be 10%
  const discounts = prodIds.map(p => 0.1)
  return of(discounts)
}

// this function simulates a call to get the tax for arrays of max 50 products
function taxes(prodIds: string[]) {
  console.log('taxes batch len', prodIds.length)
  // taxes are fixed to be 20%
  const taxes = prodIds.map(p => 0.2)
  return of(taxes)
}

// build the initial cart which contains 100 items - each item contains the product id 
// the quantity and the price (quantity and price as set to 1)
let cart: {prodId: string, qty: number, price: number}[] = new Array(100).fill(null)
cart = cart.map((_, i) => ({prodId: 'p' + i, qty: 1, price: 1}))

// this is the function that returns an Observabel which notifies the total of the order
function orderTotal$(cart: {
    prodId: string;
    qty: number;
    price: number;
}[])  { 
  // first we create a stream of items in the cart
  return from(cart).pipe(
    // for each item we call the function which returns the availability of the item
    // since the availability is calculated by an aync service, we use mergeeMap
    mergeMap(({prodId, qty, price}) => checkAvailability(prodId, qty).pipe(
      // we need to return not only the availability but also the rest of the data
      // so we pipe this map operator to enrich the data passed back
      map(available => ({available, prodId, qty, price}))
    )),
    // we filter only the available products
    filter(({available, prodId, qty}) => available),
    // buffer the items available in budders of 10
    bufferCount(10),
    // use mergeMap, as we do above, to retrieve the discount from the remote async service
    mergeMap(ar => {
      // return an object enriched wit the discound info
      return discount(ar.map(item => item.prodId)).pipe(
        map(discs => discs.map((disc, i) => ({...ar[i], disc})))
      )
    }),
    // use this merge mergeMap to tranform the emissions of batches of items with length 10
    // into a stream of 10 items
    mergeMap(ar => ar),
    // now create batched with size 50
    bufferCount(50),
    // use mergeMap to invoKe the remote service which returns the taxes
    mergeMap(ar => {
      const pIds = ar.map(item => item.prodId)
      return taxes(pIds).pipe(
        map(txs => txs.map((tax, i) => ({...ar[i], tax})))
      )
    }),
    // again use mergeMap to generare a strea  of 50 items
    mergeMap(ar => ar),
    // use reduce to calculate the order total
    reduce((orderTotal, item) => {
      const itemPrice = item.price
      orderTotal = orderTotal + (itemPrice - itemPrice * item.disc + itemPrice * item.tax) * item.qty
      return orderTotal
    }, 0)
  )
}

// subscribe to the result of orderTotal$ function to get the total of the orer
orderTotal$(cart).subscribe(d => console.log('order total', d))