Java 8 流附加错误处理供以后使用

Question

假设我有以下要重构的方法

protected Stream<T> parseFile(File file, Consumer<File> cleanup) {
  try {
    return parser.parse(file); // returns a Stream<T>
  } catch (XmlParseException e) { // child of RuntimeException
    throw new CustomRuntimeException(e);
  } finally {
    if (file != null) {
      cleanup.accept(file);
    }
  }

  throw new IllegalStateException("Should not happen");
}

此方法的目的是充当代理附加错误处理，在包装异常 CustomRuntimeException 中重新抛出流。因此，当我们稍后在流程中使用它时，我不必到处处理这些异常，而只需处理 CustomRuntimeException。

在上游，我使用了如下方法

try {
  Stream<T> stream = parseFile(someFile);
  stream.map(t -> ...);
catch (CustomRuntimeException e) {
  // do some stuff
}

下面是 parser.parse 方法的样子

public Stream<T> parse() {
  // ValueIterator<T> implements Iterator<T>, AutoCloseable
  XmlRootParser.ValueIterator<T> valueIterator = new XmlRootParser.ValueIterator(this.nodeConverter, this.reader, this.nodeLocalName, this.nodeName);
  Stream<T> stream = StreamSupport.stream(Spliterators.spliteratorUnknownSize(valueIterator, 1040), false);
  stream.onClose(valueIterator::close);
  return stream;
}

我要处理的异常将由 ValueIterator.hasNext 方法抛出。这意味着它们不会在 Stream 创建时抛出，而只会在 Stream 消耗时抛出(在流上调用 foreach/map/count/collect/...)。

如何在方法 parseFile 中很好地附加错误处理到我的流上，而不必消耗流？可能吗？

很明显，只有当 parser.parse 方法在返回它之前消耗了它的流时，这段代码才有效。这是反对使用流的。

Answer 1

提供迭代器逻辑的 Stream 后端是 Spliterator。

因此您可以使用包装器 Spliterator 来包装元素处理，如下所示:

class Wrapper<T> implements Spliterator<T> {
    final Spliterator<T> source;
    public Wrapper(Spliterator<T> source) {
        this.source = source;
    }
    @Override
    public boolean tryAdvance(Consumer<? super T> action) {
        try {
            return source.tryAdvance(action);
        }
        catch(XmlParseException ex) {
            throw new CustomRuntimeException(ex);
        }
    }
    @Override
    public void forEachRemaining(Consumer<? super T> action) {
        try {
            source.forEachRemaining(action);
        }
        catch(XmlParseException ex) {
            throw new CustomRuntimeException(ex);
        }
    }
    @Override public Spliterator<T> trySplit() {
        Spliterator<T> srcPrefix = source.trySplit();
        return srcPrefix == null? null: new Wrapper<>(srcPrefix);
    }
    @Override public long estimateSize() { return source.estimateSize(); }
    @Override public int characteristics() { return source.characteristics(); }
    @Override public Comparator<? super T> getComparator(){return source.getComparator();}
}

它保留了原始 Spliterator 的所有属性，并且只转换迭代期间抛出的异常。

然后你就可以像这样使用了

protected Stream<T> parseFile(File file) {
    Stream<T> s = parser.parse();
    return StreamSupport.stream(new Wrapper<>(s.spliterator()), s.isParallel())
                        .onClose(s::close);
}

并且调用者不应该忘记正确关闭流:

    ResultType result;
    try(Stream<T> s = parseFile(file)) {
        result = s.
         // other intermediate ops
         // terminal operation
    }

或

    ResultType result;
    try(Stream<T> s = parseFile(file)) {
        result = s.
         // other intermediate ops
         // terminal operation
    }
    finally {
        // other cleanup actions
    }