如何在 RxJava 中进行递归 Observable 调用？

Question

我对 RxJava（以及一般的 Reactive 范式）还很陌生，所以请多多包涵。

假设我有这个 News 和这个嵌套的 Comment 数据结构：

public class News {
  public int id;
  public int[] commentIds; //only top level comments
  public News(int id, int[] commentIds) {
    this.id = id;
    this.commentIds = commentIds;
  }
}

public class Comment {
  public int id;
  public int parentId; //ID of parent News or parent comment
  public int[] childIds;
  public Comment(int id, int parentId, int[] childIds) {
    this.id = id;
    this.parentId = parentId;
    this.childIds = childIds;
  }
}

假设我有这个 API 端点：

getComments(int commentId) //return Observable<Comment> for Comment with ID commentId

现在，让我们假设：

getComments(1); //will return Comment(1, 99, [3,4])
getComments(2); //will return Comment(2, 99, [5,6])
getComments(3); //will return Comment(3, 1, [])
getComments(4); //will return Comment(4, 1, [])
getComments(5); //will return Comment(5, 2, [])
getComments(6); //will return Comment(6, 2, [])

**

现在，如果我有 News n = News(99, [1,2])，我如何递归地获得它的所有子评论？即获取 ID 为 [1,2,3,4,5,6]?

的评论

**

我搜索并偶然发现了这个：https://jkschneider.github.io/blog/2014/recursive-observables-with-rxjava.html

这是递归函数：

public class FileRecursion {
    static Observable<File> listFiles(File f) {
        if(f.isDirectory())
            return Observable.from(f.listFiles()).flatMap(FileRecursion::listFiles);
        return Observable.just(f);
    }

    public static void main(String[] args) {
          Observable.just(new File("/Users/joschneider/Desktop"))
                  .flatMap(FileRecursion::listFiles)
                  .subscribe(f -> System.out.println(f.getAbsolutePath()));
    }
}

它展示了一个关于如何进行递归可观察调用的示例，但内部函数 (f.listFiles()) 是一个阻塞操作（return 不是另一个可观察对象）。在我的例子中，内部函数 (getComments) 是一个非阻塞函数，return 是另一个 Observables。我该怎么做？

任何帮助将不胜感激。

Answer 1

这实际上与文章中描述的相同：

Observable<Comment> getInnerComments(Comment comment) {
    if (comment.childIds.length > 0)
        return Observable.merge(
                Observable.just(comment),
                Observable.from(comment.childIds)
                        .flatMap(id -> getComments(id))
                        .flatMap(this::getInnerComments));
    return Observable.just(comment);
}

public static void main(String[] args) {
    getComments(1)
          .flatMap(this::getInnerComments)
          .subscribe(c -> System.out.println(comment.toString()));
}

我从 id = 1 的评论开始，然后将它传递给 getInnerComments()。 getInnerComments() 检查评论是否有 children。如果是，它会遍历每个 child id (Observable#from) 并用你的 getComments(int) API 加载每个 child。然后将每个 child 传递给 getInnerComments() 以执行相同的过程。如果评论没有 children，则立即使用 Observable#just.

返回

这是 pseudo-code，未经测试，但您应该明白了。

---

下面是一个示例，说明如何获取所有评论，然后将它们聚合为一个 List<Comment>。

getNews(99)
        .flatMap(news -> Observable.from(news.commentIds))
        .flatMap(commentId -> getComments(commentId))
        .flatMap(comment -> getInnerComments(comment))
        .toList()
        .subscribe(commentList -> { });