java - RxJava 收集() & takeUntil()

Question

我有一个未知大小的用户列表。我想要的是查询 first 30 并更新 UI。然后我想通过以 100 为步长的偏移量来查询所有其他用户，直到我得到最后一批用户 - 我应该在这里使用 takeUntil 吗？)当我得到 - 我通过添加剩余用户(结合reduce() 我相信)。

这是我的代码:

final int INITIAL_OFFSET = 0;
final int INITIAL_LIMIT = 30;
// Loading first 30 users to immediately update UI (better UX)
getServerApi().getAllFriends(userId, "photo_50", INITIAL_OFFSET, INITIAL_LIMIT)
        // Loading remaining users 100 by 100 and updating UI after all users been loaded
        .flatMap(users -> {
            AtomicInteger newOffset = new AtomicInteger(INITIAL_LIMIT);
            return Observable.just(users)
                    .flatMap(users1 -> getServerApi().getAllFriends(userId, "photo_50", newOffset.get(), Config.DEFAULT_FRIEND_REQUEST_COUNT))
                    .subscribeOn(Schedulers.io())
                    .observeOn(Schedulers.io())
                    .collect(() -> new ArrayList<User>(), (b, s) -> {
                        b.addAll(s);
                        newOffset.set(newOffset.get() + Config.DEFAULT_FRIEND_REQUEST_COUNT);
                    })
                    .repeat()
                    .takeUntil(friends -> friends.size() == 0);
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(users -> getView().appendAllFriends(users),
                throwable -> getView().setError(processFail(throwable, ServerApi.Action.GET_ALL_FRIENDS), false));

但似乎我做错了什么，因为每次进行改造调用时都会调用 onNext。

Answer 1

回答我自己的问题。阿德尔的回答很好，但我需要一个订阅(我正在使用 Nucleus MVP library )并且我想使用 collect() 和 takeUntil() 而不是 while 循环(这需要阻塞改造接口(interface)方法)。

花了几个小时终于搞定了:

final int INITIAL_LIMIT = 30;
// Loading first 30 users to immediately update UI (better UX)
getServerApi().getAllFriends(userId, "photo_50", null, INITIAL_LIMIT)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        // Updating UI 1st time or show error
        .doOnNext(users -> getView().appendAllFriends(users))
        .doOnError(throwable -> getView().setError(processFail(throwable, ServerApi.Action.GET_ALL_FRIENDS), false))
        // Loading remaining users 100 by 100 and updating UI after all users been loaded
        .flatMap(users -> {
            AtomicInteger newOffset = new AtomicInteger(INITIAL_LIMIT);
            ArrayList<User> remainingUsers = new ArrayList<>();
            AtomicBoolean hasMore = new AtomicBoolean(true);
            return Observable.just(users)
                    .flatMap(users1 -> getServerApi().getAllFriends(userId, "photo_50", newOffset.get(), Config.DEFAULT_FRIEND_REQUEST_COUNT))
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .collect(() -> remainingUsers, (b, s) -> {
                        // Needed for takeUntil
                        hasMore.set(b.addAll(s));
                        newOffset.set(newOffset.get() + Config.DEFAULT_FRIEND_REQUEST_COUNT);
                    })
                    .repeat()
                    .takeUntil(friends -> !hasMore.get())
                    // Grab all items emitted by collect()
                    .last()
                    // Updating UI last time
                    .doOnNext(users2 -> getView().appendAllFriends(users2));
        })
        .subscribe();

也许它对其他也在使用 Nucleus 的人有用。