注册流“完成”钩子

Question

使用Java 8 StreamAPI，我想按照以下方式注册“完成钩子”：

Stream<String> stream = Stream.of("a", "b", "c");

// additional filters / mappings that I don't control
stream.onComplete((Completion c) -> {
    // This is what I'd like to do:
    closeResources();

    // This might also be useful:
    Optional<Throwable> exception = c.exception();
    exception.ifPresent(e -> throw new ExceptionWrapper(e));
});

之所以要这样做，是因为我想将资源包装在Stream供API客户端使用的资源中，并且我希望Stream资源一旦被使用就自动清除。如果可能的话，客户可以致电：

Collected collectedInOneGo =
Utility.something()
       .niceLookingSQLDSL()
       .moreDSLFeatures()
       .stream()
       .filter(a -> true)
       .map(c -> c)
       .collect(collector);

而不是当前需要什么：

try (Stream<X> meh = Utility.something()
                            .niceLookingSQLDSL()
                            .moreDSLFeatures()
                            .stream()) {

    Collected collectedWithUglySyntacticDissonance =
    meh.filter(a -> true)
       .map(c -> c)
       .collect(collector);
}

理想情况下，我想了解java.util.stream.ReferencePipeline的各种方法，例如：

@Override
final void forEachWithCancel(Spliterator<P_OUT> spliterator, Sink<P_OUT> sink) {
    try {

        // Existing loop
        do { } while (!sink.cancellationRequested() && spliterator.tryAdvance(sink));
    }

    // These would be nice:
    catch (Throwable t) {
        completion.onFailure(t);
    }
    finally {
        completion.onSuccess();
    }
}

使用现有的JDK 8 API有没有简便的方法？

Answer 1

除了flatMap基于解决方案（由@Holger提出）以外，任何拦截终端操作的解决方案都将易碎于以下代码：

Stream<String> stream = getAutoCloseableStream();
if(stream.iterator().hasNext()) {
    // do something if stream is non-empty
}

根据规范，此类使用绝对合法。不要忘了，iterator()并spliterator()为终端流操作，但其执行后，你仍然需要流源的访问。同样，在任何状态下放弃Iterator或Spliterator都是完全有效的，因此您只是不知道是否会进一步使用它。

您可以考虑劝告用户不要使用iterator()和spliterator()，但是这段代码呢？

Stream<String> stream = getAutoCloseableStream();
Stream.concat(stream, Stream.of("xyz")).findFirst();

这在内部spliterator().tryAdvance()用于第一个流，然后将其放弃（如果close()显式调用生成的流，则将关闭）。您将需要询问用户也不要使用Stream.concat它。据我在您的图书馆内部所知，您经常使用iterator()/
spliterator()，因此您将需要重新访问所有这些地方以解决可能的问题。而且，当然还有许多其他库也使用iterator()/，spliterator()并且此后可能会短路：所有这些库都将与您的功能不兼容。

为什么flatMap基于解决方案的方法在这里起作用？因为在第一次调用hasNext()or时，tryAdvance()它会将 整个
流内容转储到中间缓冲区中并关闭原始流源。因此，根据流大小，您可能会浪费很多中间内存，甚至会浪费大量内存OutOfMemoryError。

您也可以考虑将PhantomReferences
保留在Stream对象上并监视ReferenceQueue。在这种情况下，完成将由垃圾收集器触发（这也有一些缺点）。

总之，我的建议是继续尝试资源。