rx-java - 使用 RxJava 重复改造调用，直到条件有效

Question

问题：点击一个端点，直到返回的项目列表为空。每个连续调用都会有一个更新的 packetId 查询参数，它是最后一项的 packetId。

设置：Retrofit2 与 Rx 适配器

尝试：

MyApiService api = retrofit.create(MyApiService.class);
Observable<Output> call = api.myCall(packetId);
call.repeat().takeUntil(output -> output != null && !output.isEmpty())
    .compose(applySchedulers())
    .subscribe(output -> {
        packetId = output.lastPacketId();
        onFetchOutput(output)
    });

这里的 packetId 不会被更新，因为 observable 将使用与它创建时相同的参数。

获取结果后，我总是可以再次调用此方法。

问题：有没有一种被动的方式来做到这一点？我假设它将涉及一些以某种方式反馈结果的操作员。

Answer 1

您可以使用PublishSubject

private static void test() {
    PublishSubject<Integer> packetSubject = PublishSubject.create();
    packetSubject
            .flatMap(integer -> call()
                    .doOnNext(output -> {
                        if (output == null) {
                            packetSubject.onCompleted();
                        }
                    }))
            .subscribe(output -> {
                if (output != null) {
                    packetSubject.onNext(output.lastPacketId);
                }});
    packetSubject.onNext(1);
}