RxJs - 创建 Observable 的 Observable 数组的正确模式/方法是什么？

Question

我对 RxJs 相当陌生，并试图让我的头脑了解正确的模式是简单地创建 Observable 的 Observable 数组。

我想检索用户帖子列表。 Posts 本身应该是 Observables，我想将它们全部保存在一个 Observable 数组中，这样当数组更改时，应该通知调用代码并更新订阅帖子“列表”的任何内容。这很简单，但我也希望每个帖子都是可观察的，所以如果我从中检索特定的帖子[i]，我也应该能够订阅这些单独的对象。 这样做的正确方法是什么？

我正在使用 Angular 9，我有：

public getPosts(): Observable<Array<Post>> {
    return new Promise((resolve, reject) = {
        let posts: Observable<Array<Post>> = new Observable<Array<Post>>();
        this.get<Array<Post>>('posts').subscribe(r => {
            posts = from(r);
            return resolve(posts);
        });
    });
}

这给了我一个Observable<Array<Post>>，但我应该如何创建一个Observable<Array<Observable<Post>>>？ 这是反模式吗？

Answer 1

这一切都是为了方便，如果您的服务器为您提供帖子中更改的差异数据，那么继续创建Observable<Observable<Post>[]>。

---

但是，在您的帖子中，存在多个问题。您不能将Observables 与Promises 混合使用。 getPosts 方法将只返回您从 API 获得的第一篇文章。

---

这是要求的解决方案，但我不确定，这是否是您真正想要的...

public getPosts(): Observable<Array<Observable<Post>>> {
  return this.get('posts').pipe(
    switchMap(posts => combineLatest(
      posts.map(post => this.get('post', post.id))
    )),
  );
}

Answer 2

目前尚不清楚您要在这里完成什么，但您可能想要更像这样的东西：

@Injectable({providedIn:'root'})
export class PostService {
  // private replay subject will cache one value
  private postSource = new ReplaySubject<Post[]>(1)
  // public list of posts observable
  posts$ = this.postSource.asObservable();
  // function to select item by id out of list
  post$ = (id) => this.posts$.pipe(map(posts => posts.find(p => p.id === id)))

  getPosts() {
    // function to get remote posts
    return this.get<Post[]>('posts');
  }

  loadPosts() {
    // function to load posts and set the subject value
    this.getPosts().subscribe(posts => this.postSource.next(posts));
  }
}

您必须定义 get 函数并在每次要更新列表时调用 loadPosts。

Answer 3

给定信息：

！如果这些陈述有任何错误，请告诉我，我会更新答案！

1. 获取函数，该函数返回一个可观察对象，其中包含一个填充了帖子的数组
2. get observable 总是在帖子发生变化时发出
3. observable (Array>) 中的值是不可观察的，不会随时间变化

this.get<Array<Post>>('posts')

可能的功能

1. () => getPostById$

// This function returns you an observable with the post related to your id.
// If the id is not found the observable will not emit
// If the id is found the observable will only emit if the interface values have been changed
function getPostById$(id: string): Observable<Post> {
  // Returns you either the post or undefined if not found
  const findId = (id: string) => (posts: Array<Post>): Post | undefined =>
     posts.find(post => post.id === id);

  // Allows you only to emit, if id has been found
  const existingPost = (post: Post | undefined): boolean => post != null;

  // Allows you only to emit if your id has been changed
  const postComparator = (prevPost: Post, currPost: Post): boolean =>
    prevPost.value === currPost.value && prevPost.name === currPost.name;

  return this.get('posts').pipe(
    map(findId(id)),
    filter(existingPost),
    distinctUntilChanged(postComparator)
  );
}

1. () => getPosts$

function getPosts$(): Observable<Array<Post>> {
  return this.get('posts');
}

1. () => getStatePosts$

// This function allows to manage your own state
// 1. posts$: overwrites all posts
// 2. clear$: empties your posts$ observable
// 3. add$: adds one observable to the end of your posts
function statePosts$(posts$: Observable<Array<Posts>>, clear$: Observable<void>, add$: Observable<Post>): Observable<Array<Post>> {
  const updatePosts = (newPosts: Array<Posts>) => (oldPosts: Array<Posts>) => newPosts;

  const clearPosts = () => (oldPosts: Array<Posts>) => [];

  const addPost = (post: Post) => (oldPosts: Array<Posts>) => [...oldPosts, post];

  return merge(
    // You can add as much update functions as you need/want (eg: deleteId, addPostAtStart, sortPosts, ...)
    posts$.pipe(map(updatePosts)),
    clear$.pipe(map(clearPosts)),
    add$.pipe(map(addPost))
  ).pipe(
    // The fn in you scan is the (oldPosts: Array<Posts>) function from one of your three update functions (updatePosts, clearPosts and addPosts).
    // Whenever one of those three observables emits it first calls the left side of the function inside the map (post: Post) and returns a new function
    // When this function reaches the scan it gets the oldPosts and is able to update it
    scan((oldPosts, fn) => fn(oldPosts), [])
  )
}

// Usage
private posts$: Observable<Array<Post>> = this.get('posts');
private clear$: Subject<void> = new Subject();
private add$: Subject<Post> = new Subject();

public statePosts$ = getStatePosts(posts$, clear$, add$);

提示： 首先尝试从 return 语句中读取函数。然后检查映射/过滤或其他操作中发生了什么。希望我没有让你太困惑。如果您有任何问题，请随时提问。