仅在需要时在Retor的Flux中请求下一个

Question

我有一个API，它返回一个限制为100个实体的实体列表。如果有更多实体，它将为下一页返回一个标记。

我想创建一个返回所有实体（所有页面）的通量，但仅在需要时（如果要求）。

我编写了以下代码：

class Page {
    String token;
    List<Object> entities;
}

Flux<Object> load(String token, final Function<String, Mono<Page>> fct) {
    return fct.apply(token).flatMapMany(page -> {
        if (page.token == null) {
            // no more pages
            return Flux.fromIterable(page.entities);
        }

        return Flux.fromIterable(page.entities).concatWith(Flux.defer(() -> load(page.token, fct)));
    });
}

而且几乎成功了

如果我请求99个元素，则加载第一页，我的流量包含99个元素。

如果我请求150个元素，则加载第一页和第二页，并且我的流量包含150个元素。

但是，如果我请求100个元素，则会加载第一页和第二页（我的流量包含100个元素）。我这里的问题是，第二个页面是在我没有请求第101个元素的地方加载的。

当前行为：

subscribe()
=> Function is called to load page 1
request(10)
=> Received: 0-9
request(89)
=> Received: 10-98
request(1)
=> Received: 99
=> Function is called to load page 2
request(1)
=> Received: 100

应为：在最后一个请求（1）之后加载第2页

这几乎像是在某个地方有预回迁，但我看不到在哪里。有什么想法吗？

Answer 1

好的，我找到了。没有按发言进行预回迁。事实上，这是通量。延迟，即在订阅时而不是在请求时加载下一页。

要解决这一问题，一个快速（且肮脏）的测试是：

Flux<Object> load(String token, final Function<String, Mono<Page>> fct) {
    return fct.apply(token).flatMapMany(page -> {
        if (page.token == null) {
            // no more pages
            return Flux.fromIterable(page.entities);
        }

        return Flux
                .fromIterable(page.entities)
                .concatWith(
                        // Flux.defer(() -> load(page.token, fct))
                        Flux.create(s -> {
                            DelegateSubscriber[] ref = new DelegateSubscriber[1];

                            s.onRequest(l -> {
                                if (ref[0] == null) {
                                    ref[0] = new DelegateSubscriber(s);
                                    load(page.token, fct).subscribe(ref[0]);
                                }
                                ref[0].request(l);
                            });
                        }));
    });
}

static class DelegateSubscriber extends BaseSubscriber<Object> {

    FluxSink<Object> delegate;

    public DelegateSubscriber(final FluxSink<Object> delegate) {
        this.delegate = delegate;
    }

    @Override
    protected void hookOnSubscribe(Subscription subscription) {
        // nothing
    }

    @Override
    protected void hookOnNext(Object value) {
        delegate.next(value);
    }

    @Override
    protected void hookOnError(Throwable throwable) {
        delegate.error(throwable);
    }
}