RxJs Пропустить пример на первом Emit, подождать, пока завершится один Observable - PullRequest
Новая
       12

RxJs Пропустить пример на первом Emit, подождать, пока завершится один Observable

0 голосов
/ 13 сентября 2018

Вопрос 1 

 combineLatest(this.layerService.layersData$, this.displayService.displayData$, this.dataSource.data$,
      (layer, display, data) => ({ layer, display, data }))
      .pipe(
        skipWhile(({ layer, display, data }) =>
          _.isEmpty(layer) || _.isEmpty(display) || _.isEmpty(data)),
        takeWhile(() => this.cacheService.isDirty()),
        sample(interval(2000)),
        map(result => {
          const layerFiltered = result.layer.filter(ly => result.display.findIndex(d => d.id === ly.id) !== -1);
          return { ...result, layer: layerFiltered };
        })
  )
  .subscribe(result => {
    console.log(result);
  });

Я хочу избежать выборки на самом первом излучении и использовать выборку после этого.

Самым первым излучением, я имею в виду, этобыл в состоянии добраться до функции карты.Может ли это быть достигнуто без использования внешней локальной переменной?

Вопрос 2 

ngOnInit() {
   this.displayService.displayData$.delay(500).take(1).subscribe(data =>  this.displayMeta = data);

   this.layerService.getLayerData()
     .subscribe(layers => {
       this.layers = layers;
     });
}

Я хочу подписаться на layerService, чтобы дождаться завершения displayService, я могу поставить layerServiceлогика подписки внутри метода подписки displayService, но это не кажется хорошим решением проблемы.

Я хочу, чтобы this.displayService ....... код был синхронным.Это также требуется один раз, а не оператор take (1).

Вопрос 3 

dirty = {};
fetchedData = {};
reportData$ = new BehaviorSubject({});

constructor(private dataSourceService: DataSourceService, private someService: SomeService) {
  const dataFetch$ = this.dataSourceService.data$
    .pipe(
      tap(dList => {
        // update dirty by comparing dList, if this.dirty has 3 keys and dList have two item then this.dirty length will be two
        this.dirty = dList.reduce((acc, et) => ({ ...acc, [et.id]: _.get(this.dirty, et.id, true) }), {});
      }),
      filter(dList => !_.isEmpty(dList)),
      map(dList => _.filter(dList, dL => this.dataSourceService.dirty[dL.id])),
      concatMap(dList => from(dList)),
      flatMap(dItem => this.someService.getDataFromApi(dItem), (item, data) => {
        return { id: item.id, data };
      }),
      tap(({ id, data }) => {
        this.fetchedData[id] = data;
        this.dirty[id] = false;
        this.dataSourceService.resetDirty(id);
      })
    );

  dataFetch$.merge(this.dataSourceService.data$)
    .subscribe(() => {
      this.fetchedData = _.pickBy(this.fetchedData, (__, key) => _.has(this.dirty, key));
      this.reportData$.next(this.fetchedData);
    });
}

Метод подписки должен вызываться, даже если фильтрвернуть ложьПроблема с вышеуказанным подходом состоит в том, что подписка будет вызываться дважды.

Если dList пуст, dataFetch $ не вызывается, поэтому подписка вызывается один раз, но если она не пуста, то подписка вызывается дважды.

Конструкция такова: если элемент удаляется из this.dataSourceService.data $ один за другим и, наконец, this.dataSourceService.data $ .length становится 0, наблюдаемая цепочка не достигнет подписки, в этом случае также сделайте это.fetchedData = empty

Поскольку элемент из dataSourceService.data $ удаляется, соответствующий элемент из this.fetchedData должен быть удален, я не знаю, какой элемент удален, поэтому грязный флаг, обратите внимание на первую операцию касания.В подписке dirtyList используется для обновления fetchedData.

1 Ответ

0 голосов
/ 14 сентября 2018

Вопрос 1

Возможно, вы захотите создать Observable, похожий на тот, который у вас уже есть, но без оператора sample.Если у вас есть такая наблюдаемая, у вас есть базовый строительный блок, который позволяет вам получить то, что вы хотите получить.

Базовая наблюдаемая должна выглядеть примерно так: 

const basicObs = combineLatest(this.layerService.layersData$, this.displayService.displayData$, this.dataSource.data$,
      (layer, display, data) => ({ layer, display, data }))
      .pipe(
        skipWhile(({ layer, display, data }) =>
          _.isEmpty(layer) || _.isEmpty(display) || _.isEmpty(data)),
        takeWhile(() => this.cacheService.isDirty()),
        map(result => {
          const layerFiltered = result.layer.filter(ly => result.display.findIndex(d => d.id === ly.id) !== -1);
          return { ...result, layer: layerFiltered };
        })
  )

Тогда вы можете простообъедините первый выброс basicObs со всеми последующими выбросами, используя оператор concat.Код будет выглядеть так: 

const firstNotification = basicObs.pipe(
  take(1)
);
const followingNotifications = basicObs.pipe(
  skip(1), // to avoid emitting the first element
  sample(interval(2000))
);

firstNotification.pipe(
  concat(followingNotifications)
)
.subscribe(result => console.log(result))

Вопрос 2

Если вы хотите, чтобы подписка на this.displayService.displayData$.delay(500).take(1) была выполнена до того, как подписка на this.layerService.getLayerData() будет выполнена,тогда вы можете попробовать что-то вроде этого 

ngOnInit() {
   this.displayService.displayData$.pipe(
      delay(500),
      take(1),
      tap(data =>  this.displayMeta = data),  // this is the side effect that you have with the first subscription
      switchMap(() => this.layerService.getLayerData())
   )
     .subscribe(layers => {
       this.layers = layers;
     });
}

Ключевая идея здесь - использовать switchMap для переключения с первого Observable, как только оно выдает, на второе.Перед переключением через оператора tap мы запускаем побочный эффект, который был встроен в подписку this.displayService.displayData$.delay(500).take(1).

...