java - 满足特定条件时结束 "infinite"流

Question

我正在尝试从以页面形式提供内容的 REST 风格网络服务中提取数据。

我知道我已经到达终点的唯一方法是当我请求一个页面但没有结果时。我想在那个时候终止流。

我编写了以下 Java 代码。第一个函数从 Web 服务中拉取单个页面并将其作为流返回。第二个函数将流平面映射到一个流中。

public Stream<ApplicationResponse> getApplications(String token, RestTemplate rt, Integer page, Integer pageSize) {
    HttpEntity<String> entity = new HttpEntity<>("parameters", getHeaders(token));
    String url = String.format("%s?PageIndex=%s&PageSize=%s", endpoint, page, pageSize);
    ResponseEntity<ApplicationCollection> ar = rt.exchange(url, HttpMethod.GET, entity, ApplicationCollection.class);
    ApplicationResponse[] res = Objects.requireNonNull(ar.getBody()).getData();

    // Do something here when res is empty, so that the stream ends

    return Arrays.stream(res);
}

public Stream<ApplicationResponse> getApplications(String token, RestTemplate rt) {
    // This function does the right thing, exept when we run out of data!
    return IntStream.iterate(1, i -> i + 1).mapToObj(i -> getApplications(token, rt, i, 500)).flatMap(Function.identity());
}

问题是，我该如何让这一切结束？

如果我用 Python 编写此代码，我会在我知道没有任何内容可放入流时引发 StopIteration 异常。我可以做类似的事情吗？

我能想到的最好的办法是使用 null，或者引发异常来表示数据结束，然后将流包装到迭代器中，该迭代器知道在收到该信号时停止。但是我还能做些更符合惯用语的事情吗？

Answer 1

在 Holger 的评论之后，我试了一下并尝试使用 Spliterator 而不是 Iterator。它确实更简单，因为 next 和 hasNext 是......有点组合成 tryAdvance？它甚至足够短，只需将它内联到一个 util 方法中，imo。

public static Stream<ApplicationResponse> getApplications(String token, RestTemplate rt)
{
    return StreamSupport.stream(new AbstractSpliterator<ApplicationResponse[]>(Long.MAX_VALUE,
                                                                               Spliterator.ORDERED
                                                                                               | Spliterator.IMMUTABLE)
    {
        private int page = 1;

        @Override
        public boolean tryAdvance(Consumer<? super ApplicationResponse[]> action)
        {
            HttpEntity<String> entity = new HttpEntity<>("parameters", getHeaders(token));
            String url = String.format("%s?PageIndex=%s&PageSize=%s", endpoint, page, 500);
            ResponseEntity<ApplicationCollection> ar = rt.exchange(url, HttpMethod.GET, entity,
                                                                   ApplicationCollection.class);
            ApplicationResponse[] res = Objects.requireNonNull(ar.getBody()).getData();

            if (res.length == 0)
                return false;

            page++;
            action.accept(res);
            return true;
        }
    }, false).flatMap(Arrays::stream);
}

---

您可以实现一个 Iterator 并创建一个 Stream:

public class ResponseIterator
    implements Iterator<Stream<ApplicationResponse>>
{
    private int                   page = 1;
    private String                token;
    private RestTemplate          rt;

    private ApplicationResponse[] next;

    private ResponseIterator(String token, RestTemplate rt)
    {
        this.token = token;
        this.rt = rt;
    }

    public static Stream<ApplicationResponse> getApplications(String token, RestTemplate rt)
    {
        Iterable<Stream<ApplicationResponse>> iterable = () -> new ResponseIterator(token, rt);
        return StreamSupport.stream(iterable.spliterator(), false).flatMap(Function.identity());
    }

    @Override
    public boolean hasNext()
    {
        if (next == null)
        {
            next = getNext();
        }
        return next.length != 0;
    }

    @Override
    public Stream<ApplicationResponse> next()
    {
        if (next == null)
        {
            next = getNext();
        }
        Stream<ApplicationResponse> nextStream = Arrays.stream(next);
        next = getNext();
        return nextStream;
    }

    private ApplicationResponse[] getNext()
    {
        HttpEntity<String> entity = new HttpEntity<>("parameters", getHeaders(token));
        String url = String.format("%s?PageIndex=%s&PageSize=%s", endpoint, page, 500);
        ResponseEntity<ApplicationCollection> ar = rt.exchange(url, HttpMethod.GET, entity,
                                                               ApplicationCollection.class);
        ApplicationResponse[] res = Objects.requireNonNull(ar.getBody()).getData();
        page++;
        return res;
    }
}

它将在hasNext() 中检查下一个响应是否为空，从而停止流。否则，它将对该响应进行流式处理和平面映射。我已经硬连接了 pageSize，但您可以轻松地将其作为工厂方法 ResponseIterator.getApplications() 的第三个输入。