Есть ли способ объединить дочернее свойство обратно в родительский источник после некоторых операций в RxJs?

Question

У меня есть оператор, который выполняет некоторые рекурсивные операции над дочерним свойством источника. Что мне делать, чтобы объединить дочернее свойство обратно в источник после того, как я выполнил рекурсивную операцию?

const state = {
  posts: [
    {id: 3, title: 't1', userId: 1},
  ],
  index: 0,
  config: {
    previousBufferSize: 1,
    nextBufferSize: 1,
  }
};
const source = new BehaviorSubject(state);

const generatePreviousPosts$ = (posts) => {
  return Observable.create(observer => {
    getPost(posts[0].id - 1)
    .then(previousPost => {
      observer.next([previousPost, ...posts]);
    });
  });
};

const previousBuffer$ = source.pipe(
    pluck('posts'),
    expand(generatePreviousPosts$),
    tap(console.log),
    // What do I do to merge post back in the state so I could use takeWhile?
    takeWhile(state => {
      const {posts, config, index} = state;
      return posts.length <= config.previousBufferSize - index + posts.length &&
          posts[0].id != null;
    })
);

Answer 1

Гораздо более элегантное решение, данное https://github.com/Dorus для rxjs gitter.

const shouldGetPost = ({posts, config, index}) => posts.length <= config.previousBufferSize - index + posts.length
      && posts[0].id != null

const generatePreviousPosts = ({posts, config, index}) => !shouldGetPost({posts, config, index}) ?  EMPTY : 
  from(getPost(posts[0].id - 1)).pipe(
    map(previousPost => ({[previousPost, ...posts], config, index}))
  )

const previousBuffer$ = source.pipe(
    expand(generatePreviousPosts)
  );

Answer 2

Один из способов сделать это - использовать mergeMap, но я чувствую, что может быть более элегантное решение этой проблемы.

const previousBuffer$ = source.pipe(
    mergeMap(
        state => (of(state.posts)
        .pipe(expand(generatePreviousPosts$))),
        (state, posts) => ({...state, posts})),
    takeWhile(state => {
      const {posts, config, index} = state;
      return posts.length <= config.previousBufferSize - index + posts.length &&
          posts[0].id != null;
    })
);