获得以前的可观察结果，以加入管道RXJS

Question

第一个HTTP调用以获取发货数据 如果发货数据不可用 将停止显示错误

如果发货数据可用我需要致电第二和第三个呼叫paralalley

我使用叉子加入来调用paralell请求，并在获取叉子加入数据之后如何在叉子上加入地图方法中使用装运数据。

   from(getShipment(this.data.CardId))
              .pipe(
                filter(shipmentData => shipmentData != undefined),
                tap(shipmentData => {
                  if (shipmentData.ShipmentId === null || shipmentData.ShipmentId === "") {
                    throw "Shipment needs to be connected"
                  }
                }),
                //I NEED TO PASS THIS SHIPMENT DATA TO FORK JOIN PIPE METHOD
                map(shipmentData => shipmentData),                
                concatMap(shipmentData =>
                  forkJoin([
                    getOrderLineAllocationByCardIdObservable(this.data.CardId),
                    getLotsByCardIdObservable(this.data.CardId)
                  ])
                    .pipe(
                      filter(response => response != undefined),
                      map(([inventoryData, lotsData],shipmentData) => {
                                               
                       //I NEED TO ACCESS SHIPMENT DATA IN HERE
                       //SHIPMENT DATA NOT AVAILABLE IN HERE
        
                      })
                    )
                ),
                // Catch errors thrown above
                catchError(error => {                      
                  return EMPTY;
                }),
                // Always finish by Hiding Loading indicator
                finalize(() => this.store.dispatch(setLoadingSpinner({showLoading: false})))
              )
              .subscribe();

Answer 1

只需删除shipmentData作为您的地图功能的参数，它已经在那里：

import {
  of,
  map,
  Observable,
  from,
  filter,
  tap,
  concatMap,
  forkJoin,
  catchError,
  finalize,
} from 'rxjs';

// Open the console in the bottom right to see results.
function getShipment(x) {
  return new Promise(function (res, rej) {
    res({ key: 'hi, im shipmentData', ShipmentId: 989 });
  });
}

function getOrderLineAllocationByCardIdObservable(x) {
  return of(1);
}

function getLotsByCardIdObservable(x) {
  return of(2);
}

from(getShipment(1))
  .pipe(
    filter((shipmentData) => shipmentData != undefined),
    tap((shipmentData: any) => {
      if (shipmentData.ShipmentId === null || shipmentData.ShipmentId === '') {
        throw 'Shipment needs to be connected';
      }
    }),
    concatMap((shipmentData) =>
      forkJoin(
        getOrderLineAllocationByCardIdObservable(666),
        getLotsByCardIdObservable(666)
      ).pipe(
        filter((response) => response != undefined),
        map(([inventoryData, lotsData]) => {
          console.log(shipmentData);
        })
      )
    ),
    // Catch errors thrown above
    catchError((error) => {
      return of();
    }),
    // Always finish by Hiding Loading indicator
    finalize(() =>
      // this.store.dispatch(setLoadingSpinner({ showLoading: false }))
      console.log('rdy')
    )
  )
  .subscribe();

可运行的示例：
https://stackblitz.com/edit/edit/redit/rxjs-t5mzxy?file= index。 TS