rxjs - 在 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;
    })
);

标签： rxjsreactive-programming

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)
  );