在 RxJava 2.x 中使用 flatMap答案

【问题标题】：Using flatMap in RxJava 2.x在 RxJava 2.x 中使用 flatMap
【发布时间】：2018-11-22 10:25:42
【问题描述】：

我正在使用自己的 API，我希望使用 RxJava 链接一些分页结果。我使用基于光标的分页。（假设第一个请求中有 50 个用户）：

{
    "data":{
        "status":"ok",
        "total":988, //users total
        "has_next_page":true,
        "end_cursor":"AQAxd8QPGHum7LSDz8DnwIh7yHJDM22nEjd",
        "users":[{"id":"91273813",
                "username":"codergirl",
                "full_name":"Code Girl",
                "picture_url":"https://cdn.com/21603182_7904715668509949952_n.jpg",
                },
                ...
                ]
        }
}

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

public class DataResponse {
    @SerializedName("end_cursor")
    private String end_cursor;

    @SerializedName("users")
    private JsonArray users;

    @SerializedName("has_next_page")
    private Boolean has_next_page;

    public boolean hasNextCursor(){
        return has_next_page;
    }
    public String endCursor(){
        if (hasNextCursor()){
            return end_cursor;
        }
        return "";
    }
    public JsonArray getUsers(){
        return users;
    }
}

然后：

public interface MyService  {
    @GET( "/users")
    Observable<DataResponse> getUsers(
            @Query("cursor") String cursor,
    );
}

和

MyService service = RetrofitClient.getInstance();
service.getUsers()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe( val->  showUsers(val.getUsers())); // getting the first 50 users

下一个调用应该是“/users?cursor=AQAxd8QPGHum7LSDz8DnwIh7yHJDM22nEjd”

我想返回所有（在本例中为 988 个）用户

【问题讨论】：

看起来您的改造服务定义不正确（或完整），即没有光标就没有 getUsers。
我写了一个决议.. 它有效，但我正在寻找更好的方法来做到这一点.. 请看我的答案
我遇到的问题是您粘贴的改造服务定义不正确。你能纠正一下吗？

标签： java pagination retrofit rx-java2 flatmap

【解决方案1】：

我的解决方案

import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

public void getAllUsers(){

    AtomicReference<String> cache = new AtomicReference<>();
    AtomicBoolean hasMore = new AtomicBoolean(true);

    io.reactivex.Observable.just(0)
        // getting the first 50 users
        .flatMap(users1-> service.getUsers( cache.get() ))
        
        // scheduler
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        
        // re-call variable
        .repeatUntil(() -> !hasMore.get())
        
        .subscribe(new Observer<DataResponse>() {
            @Override
            public void onSubscribe(Disposable d) { // on subscribe }

            @Override
            public void onNext(DataResponse response) {
                
                // saving boolean (If there are more users)
                hasMore.set(response.hasNextCursor());
                
                // saving next cursor
                cache.set(response.endCursor());
                
                // adding the new 50 users
                addToList(response.getUsers());
                
            }

            @Override
            public void onError(Throwable e) { /*error */ }

            @Override
            public void onComplete() { /*complete*/ }
        });

}

【讨论】：