RxJava 按需加载项目答案

【问题标题】：RxJava Load items on demandRxJava 按需加载项目
【发布时间】：2017-04-12 08:21:11
【问题描述】：

我想在其他一些 observable 发出新项目（触发器）时加载下一个项目。

这是发出项目的代码：

public Observable<Item> get() {
    return idApi.get().flatMap(new Function<List<String>, ObservableSource<String>>() {
        @Override
        public ObservableSource<String> apply(@NonNull List<String> ids) throws Exception {
            return Observable.fromIterable(ids);
        }
    }).flatMap(new Function<String, ObservableSource<Item>>() {
        @Override
        public ObservableSource<Item> apply(@NonNull final String id) throws Exception {
            return dataApi.get(id).map(new Function<Data, Item>() {
                @Override
                public Item apply(@NonNull Data data) throws Exception {
                    return new Item(data , id);
            });
        }
    });
}

触发 Observable：

RxView.clicks(view.findViewById(R.id.button_more)).debounce(500, TimeUnit.MILLISECONDS);

我可以解决这个问题的唯一方法是使用 Subject 并持有对不优雅且似乎没有反应性的 id 列表的引用。

编辑：这是我目前的解决方案，但我必须直接订阅触发事件。我不认为它很优雅。

@Override
public Observable<Item> get(final Observable<Object> trigger) {
    final PublishSubject<Item> subject = PublishSubject.create();
    return idApi.get().flatMap(new Function<List<String>, ObservableSource<Queue<String>>>() {
        @Override
        public ObservableSource<Queue<String>> apply(@NonNull List<String> ids) throws Exception {
            final Queue<String> q = new LinkedList<>(ids);
            return Observable.just(q);
        }
    }).flatMap(new Function<Queue<String>, ObservableSource<Item>>() {
        @Override
        public ObservableSource<Item> apply(@NonNull final Queue<String> ids) throws Exception {
            trigger.subscribeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Object>() {
                @Override
                public void accept(@NonNull Object o) throws Exception {
                    if (ids.size() > 0) {
                        final String id = ids.poll();
                        dataApi.get(id).map(new Function<Data, Item>() {
                            @Override
                            public Item apply(@NonNull Data data) throws Exception {
                                return new Item(data, id) l
                            }
                        }).subscribe(new Consumer<Item>() {
                            @Override
                            public void accept(@NonNull Item item) throws Exception {
                                subject.onNext(item);
                            }
                        });
                    } else {
                        subject.onComplete();

                    }
                }
            });
            return subject;
        }
    });
}

【问题讨论】：

为您的 clickObservable 使用平面地图。
@PhoenixWang 这就是我使用Subject 方法所做的事情，问题是如何在触发器发出新项目时从idApi 获取n 的发出计数。跨度>
所以你从 api 加载一个 id 列表，并且你想在单击按钮时为下一个 id 加载数据？或者你想用数据再次加载 id？
@Lamorak 是的，这就是我想要做的。我不想再次加载 ID。
@Pedram 您可以通过平面图传递该 ID。像 RxView.clicks(view).flatmap(ignore -> Observable.defer(get()));

标签： java android rx-java rx-java2

【解决方案1】：

使用压缩包

public Observable<Item> get(View v) {
    return idApi.get().flatMap(new Function<List<String>, ObservableSource<String>>() {
        @Override
        public ObservableSource<String> apply(@NonNull List<String> ids) throws Exception {
            return Observable.fromIterable(ids);
        }
    }).zipWith(RxView.clicks(v).debounce(500, TimeUnit.MILLISECONDS), (n, i) -> n))    
    .flatMap(new Function<String, ObservableSource<Item>>() {
        @Override
        public ObservableSource<Item> apply(@NonNull final String id) throws Exception {
            return dataApi.get(id).map(new Function<Data, Item>() {
                @Override
                public Item apply(@NonNull Data data) throws Exception {
                    return new Item(data , id);
            });
        }
    });
}

每次点击获得 N 个项目

public Observable<Item> getN(View v, int nitems) {
    return idApi.get().flatMap(new Function<List<String>, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull List<String> ids) throws Exception {
                return Observable.fromIterable(ids);
            }
        }).buffer(nitems).zipWith(RxView.clicks(v).debounce(500, TimeUnit.MILLISECONDS), (n, i) -> n))
        .flatMap(new Function<List<String>, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull final List<String> ids) throws Exception {
                return Observable.from(ids)
            }
         }
        )  
        .flatMap(new Function<String, ObservableSource<Item>>() {
            @Override
            public ObservableSource<Item> apply(@NonNull final String id) throws Exception {
                return dataApi.get(id).map(new Function<Data, Item>() {
                    @Override
                    public Item apply(@NonNull Data data) throws Exception {
                        return new Item(data , id);
                 });
               }
         }
        });
 }

编辑：您仍然必须使用 subscribeOn 来确保您在 RXView.clicks 的主线程上和任何网络的 IO 线程上。

【讨论】：

谢谢，这确实有效，但是我从RxBinding 收到错误，因为它不在main thread 上，如果我将其更改为主线程，我会因为network on main thread 而收到错误。我必须在网络调用之前添加subscribeOn(Schedulers.io())，在 zipWith 之后添加subscribeOn(AndroidScheduler.mainThread())。还有其他选择吗？
如果我想加载n 项目而不是每个触发器一个呢？
我发现缓冲区可以做到这一点，但它需要另一个平面图，这有点多余。接受这个作为答案再次感谢。

【解决方案2】：

不太好，但很有效：

你的get方法：

Observable<Item> get(final String id){
    return Observable.defer(() -> {
    dataApi.get(id).map(new Function<Data, Item>() {
            @Override
            public Item apply(@NonNull Data data) throws Exception {
                return new Item(data , id);
        });
    })
}

您的点击：

private List<String> ids = {//your id s}
private int current = 0;

RxView.clicks(view).flatmap(ignore -> get(ids.get(current++)))
                   .subscribe(//your Observer)

我想推荐zipwith() 的答案，似乎比我的答案更好。

【讨论】：