RxJS с приставкой - испускание действия по окончании (с помощью mergeMap)

Question

Это мой текущий эпос, который я написал:

const storiesURL = 'https://hacker-news.firebaseio.com/v0/topstories.json?print=pretty';
const topStoryURL = id => `https://hacker-news.firebaseio.com/v0/item/${id}.json?print=pretty`;
const maxAtOnce = 15;

const fetchStoriesEpic = action$ => action$.pipe(
  ofType(actions.FETCH_STORIES),
  debounceTime(100),
  switchMap(() => ajax.getJSON(storiesURL)), // 1. After that I have an array of ids
  map(ids => ids.slice(0, maxAtOnce).map(topStoryURL)), // 2. After that I have an array of 15 of the latest ids
  mergeMap(urls => from(urls)), // 3. After that I have those URLs
  mergeMap(url => ajax.getJSON(url), null, 3), // 4. After that I have those responses
  scan((acc, val) => [...acc, val] , []), // 5. Here I accumulate the responses into an array of stories
  map(stories => actions.fetchStoriesFullfilled(stories)), // 6. Here I dispatch an action, so it passes those stories once at a time as they arrive
  // ignoreElements(),
);

На FETCH_STORIES я делаю свой атрибут loading внутри состояния равным true. Я хотел бы установить значение false, когда эти запросы завершаются, но не после одного из них, а после того, как они завершают все (в данном случае 15 запросов).

Как мне этого добиться?

Кстати - это обычная модель? Знаете ли вы какие-либо ресурсы, где я могу найти шаблоны RxJS для асинхронных действий (которые я на самом деле буду использовать)?

Answer 1

Я думаю, вы можете упростить ваш поток. Например:

const { of, forkJoin, concat } = rxjs;
const { switchMap, map } = rxjs.operators;
const { ajax } = rxjs.ajax;

/**
 * test action creators
 */
const actions = {
  fetchStoriesStartLoading: () => ({type: 'fetchStoriesStartLoading'}),
  fetchStoriesDoneLoading: () => ({type: 'fetchStoriesDoneLoading'}),
  fetchStoriesFulfilled: (stories) => ({type: 'fetchStoriesFulfilled', stories}),
};

/**
 * configuration
 */
const storiesURL = 'https://hacker-news.firebaseio.com/v0/topstories.json?print=pretty';
const topStoryURL = id => `https://hacker-news.firebaseio.com/v0/item/${id}.json?print=pretty`;
const maxAtOnce = 15;

/**
 * this stream loads configured number of stories
 */
const stories$ = ajax.getJSON(storiesURL).pipe(
  // switch to loading individual requests, forkJoin will wait until all are resolved
  switchMap(ids => forkJoin(
    // use only max number of ids
    ids.slice(0, maxAtOnce)
      // and map those to requests
      .map(id => ajax.getJSON(topStoryURL(id))))),
);

/**
 * this stream wraps the stories$ stream with loading/fulfilled indicators
 */
const storiesWithLoadingIndicator$ = concat(
  // signal start loading
  of(actions.fetchStoriesStartLoading()),

  // load stories
  stories$.pipe(
    // signal fulfilled
    map(actions.fetchStoriesFulfilled)
  ),

  // signal done loading
  of(actions.fetchStoriesDoneLoading()),
);

/**
 * test
 */
storiesWithLoadingIndicator$.subscribe(console.log);

/**
 * as an epic
 */
const fetchStoriesEpic = action$ => action$.pipe(
  ofType(actions.FETCH_STORIES),
  debounceTime(100),
  switchMapTo(storiesWithLoadingIndicator$)
);

<script src="https://unpkg.com/rxjs@6.2.1/bundles/rxjs.umd.min.js"></script>

Answer 2

Не проверено, но я думаю, что должно работать следующее:

const fetchStoriesEpic = action$ => action$.pipe(
  ofType(actions.FETCH_STORIES),
  debounceTime(100),
  switchMap(() => ajax.getJSON(storiesURL)),
  map(ids => ids.slice(0, maxAtOnce).map(topStoryURL)),
  mergeMap(urls => from(urls)),
  mergeMap(url => ajax.getJSON(url), null, 3).pipe(
    // instead of scan use reduce to wait for completion
    reduce((acc, val) => [...acc, val] , []),
  )
  // once they're all done, then return
  map(stories => actions.fetchStoriesFullfilled(stories)),
);