RxJava 和基于游标的 RESTful 分页

Question

我正在使用 Spotify API，并希望使用 RxJava 链接一些分页结果。 Spotify 使用基于光标的分页，因此the one from @lopar 之类的解决方案将不起作用。

回复来自this call，看起来像这样（假设有50个items）：

{
  "artists" : {
    "items" : [ {
      "id" : "6liAMWkVf5LH7YR9yfFy1Y",
      "name" : "Portishead",
      "type" : "artist"
    }],
    "next" : "https://api.spotify.com/v1/me/following?type=artist&after=6liAMWkVf5LH7YR9yfFy1Y&limit=50",
    "total" : 119,
    "cursors" : {
      "after" : "6liAMWkVf5LH7YR9yfFy1Y"
    },
    "limit" : 50,
    "href" : "https://api.spotify.com/v1/me/following?type=artist&limit=50"
  }
}

现在，我正在使用改造获得前 50 个这样的结果：

public class CursorPager<T> {
    public String href;
    public List<T> items;
    public int limit;
    public String next;
    public Cursor cursors;
    public int total;

    public CursorPager() {
    }
}

public class ArtistsCursorPager {
    public CursorPager<Artist> artists;

    public ArtistsCursorPager() {
    }
}

然后

public interface SpotifyService  {

    @GET("/me/following?type=artist")
    Observable<ArtistsCursorPager> getFollowedArtists(@Query("limit") int limit);

    @GET("/me/following?type=artist")
    Observable<ArtistsCursorPager> getFollowedArtists(@Query("limit") int limit, @Query("after") String spotifyId);

}

和

mSpotifyService.getFollowedArtists(50)
        .flatMap(result -> Observable.from(result.artists.items))
        .flatMap(this::responseToArtist)
        .sorted()
        .toList()
        .subscribe(new Subscriber<List<Artist>>() {
            @Override
            public void onNext(List<Artist> artists) {
                callback.onSuccess(artists);
            }
            // ...
        });

我想返回callback.success(List<Artist>) 中的所有（在本例中为 119 位）艺术家。我是 RxJava 新手，所以我不确定是否有 智能 方法可以做到这一点。

Answer 1

递归解决方案的唯一问题是堆栈溢出问题。一种无需递归的方法是

Observable<ArtistsCursorPager> allPages = Observable.defer(() ->
{
    BehaviorSubject<Object> pagecontrol = BehaviorSubject.create("start");
    Observable<ArtistsCursorPager> ret = pageControl.asObservable().concatMap(aKey ->
    {
        if (aKey != null && aKey.equals("start")) {
            return Observable.getFollowedArtists(50).doOnNext(page -> pagecontrol.onNext(page.cursors.after));
        } else if (aKey != null && !aKey.equals("")) {
            return Observable.getFollowedArtists(50,aKey).doOnNext(page -> pagecontrol.onNext(page.cursors.after));
        } else {
            return Observable.<ArtistsCursorPager>empty().doOnCompleted(()->pagecontrol.onCompleted());
        }        
    });
    return ret;
});

查看this question的解决方案。

Answer 2

没有唯一的方法可以做到这一点。 就我而言，我所做的是使用 mergeWith 进行一些递归调用

private Observable<String> getUUIDsQuery(JsonObject response) {
    final Observable<String> uuidsQuery = createUuidsQuery(response);
    return hasPagination(response) ? paginate(response, uuidsQuery) : uuidsQuery;
}

   private Observable<String> paginate(JsonObject response, Observable<String> uuidsQuery) {
    return request(getPaginationUri(response))
            .flatMap(res -> uuidsQuery.mergeWith(getUUIDsQuery(res)));
}

希望能帮助你给你一个想法。

Answer 3

感谢您睁开眼睛，我没有正确阅读您的问题。这是我可以建议您不使用普通递归函数但使用 RxJava 方式的最佳解决方案。

PublishSubject<Integer> limit = PublishSubject.create();

        limit.concatMap(integer -> Observable.just(integer + 1)) // Assuming this gives network result based upon the artist id you provided
                .doOnNext(integer -> {
                    // Based on the network result i make my conditions here. Pass the 50th artist id here. otherwise call onCompleted.
                    Timber.d("Publish: doOnNext: %d", integer);
                    if (integer < 10) {
                        // Pass the artist id to make the next call
                        limit.onNext(integer);
                    } else {
                        limit.onCompleted();
                    }
                })
                .toList()
                .subscribe(integers -> Timber.d("Publish: All the results"));

        limit.onNext(1); // For demo, I'm starting here with 1. You have to pass the artist id here

输出如下所示：

Publish: doOnNext: 2       
Publish: doOnNext: 3       
Publish: doOnNext: 4       
Publish: doOnNext: 5       
Publish: doOnNext: 6       
Publish: doOnNext: 7       
Publish: doOnNext: 8       
Publish: doOnNext: 9       
Publish: doOnNext: 10      
Publish: All the results

toList() 操作员会在所有呼叫完成后为您提供最终得到的所有响应的列表。看看reduce() 运算符。