在RxJava 2.x中使用flatMap

Question

我在使用自己的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",
                },
                ...
                ]
        }
}

现在，我正在使用retrofit获取前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）。

- sandes

---

你的Retrofit服务定义似乎不正确（或不完整），即没有带有游标的“getUsers”。 - JohnWowUs

我写了一个解决方案。它能够工作，但我正在寻找更好的方法来完成它。请查看我的答案。 - sandes

我遇到的问题是你粘贴的Retrofit服务定义不正确。你能纠正一下吗？ - JohnWowUs

Answer 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*/ }
        });

}

- sandes