Mercurial > hg > openjdk > aarch64-port > jdk
changeset 8022:2e8d51a5596b
8017513: Support for closeable streams
8022237: j.u.s.BaseStream.onClose() has an issue in implementation or requires spec clarification
8022572: Same exception instances thrown from j.u.stream.Stream.onClose() handlers are not listed as suppressed
Summary: BaseStream implements AutoCloseable; Remove CloseableStream and DelegatingStream
Reviewed-by: alanb, mduigou, psandoz
Contributed-by: brian.goetz@oracle.com
line wrap: on
line diff
--- a/src/share/classes/java/nio/file/Files.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/nio/file/Files.java Tue Sep 03 12:16:01 2013 -0700 @@ -25,34 +25,56 @@ package java.nio.file; -import java.nio.file.attribute.*; -import java.nio.file.spi.FileSystemProvider; -import java.nio.file.spi.FileTypeDetector; +import java.io.BufferedReader; +import java.io.BufferedWriter; +import java.io.Closeable; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.io.OutputStreamWriter; +import java.io.Reader; +import java.io.UncheckedIOException; +import java.io.Writer; import java.nio.channels.Channels; import java.nio.channels.FileChannel; import java.nio.channels.SeekableByteChannel; -import java.io.Closeable; -import java.io.InputStream; -import java.io.OutputStream; -import java.io.Reader; -import java.io.Writer; -import java.io.BufferedReader; -import java.io.BufferedWriter; -import java.io.InputStreamReader; -import java.io.OutputStreamWriter; -import java.io.IOException; -import java.io.UncheckedIOException; -import java.util.*; -import java.util.function.BiPredicate; -import java.util.stream.CloseableStream; -import java.util.stream.DelegatingStream; -import java.util.stream.Stream; -import java.util.stream.StreamSupport; -import java.security.AccessController; -import java.security.PrivilegedAction; import java.nio.charset.Charset; import java.nio.charset.CharsetDecoder; import java.nio.charset.CharsetEncoder; +import java.nio.file.attribute.BasicFileAttributeView; +import java.nio.file.attribute.BasicFileAttributes; +import java.nio.file.attribute.DosFileAttributes; +import java.nio.file.attribute.FileAttribute; +import java.nio.file.attribute.FileAttributeView; +import java.nio.file.attribute.FileOwnerAttributeView; +import java.nio.file.attribute.FileStoreAttributeView; +import java.nio.file.attribute.FileTime; +import java.nio.file.attribute.PosixFileAttributeView; +import java.nio.file.attribute.PosixFileAttributes; +import java.nio.file.attribute.PosixFilePermission; +import java.nio.file.attribute.UserPrincipal; +import java.nio.file.spi.FileSystemProvider; +import java.nio.file.spi.FileTypeDetector; +import java.security.AccessController; +import java.security.PrivilegedAction; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.ServiceLoader; +import java.util.Set; +import java.util.Spliterator; +import java.util.Spliterators; +import java.util.function.BiPredicate; +import java.util.stream.Stream; +import java.util.stream.StreamSupport; /** * This class consists exclusively of static methods that operate on files, @@ -74,6 +96,21 @@ return path.getFileSystem().provider(); } + /** + * Convert a Closeable to a Runnable by converting checked IOException + * to UncheckedIOException + */ + private static Runnable asUncheckedRunnable(Closeable c) { + return () -> { + try { + c.close(); + } + catch (IOException e) { + throw new UncheckedIOException(e); + } + }; + } + // -- File contents -- /** @@ -3228,29 +3265,7 @@ // -- Stream APIs -- /** - * Implementation of CloseableStream - */ - private static class DelegatingCloseableStream<T> extends DelegatingStream<T> - implements CloseableStream<T> - { - private final Closeable closeable; - - DelegatingCloseableStream(Closeable c, Stream<T> delegate) { - super(delegate); - this.closeable = c; - } - - public void close() { - try { - closeable.close(); - } catch (IOException ex) { - throw new UncheckedIOException(ex); - } - } - } - - /** - * Return a lazily populated {@code CloseableStream}, the elements of + * Return a lazily populated {@code Stream}, the elements of * which are the entries in the directory. The listing is not recursive. * * <p> The elements of the stream are {@link Path} objects that are @@ -3264,10 +3279,13 @@ * reflect updates to the directory that occur after returning from this * method. * - * <p> When not using the try-with-resources construct, then the stream's - * {@link CloseableStream#close close} method should be invoked after the - * operation is completed so as to free any resources held for the open - * directory. Operating on a closed stream behaves as if the end of stream + * <p> The returned stream encapsulates a {@link DirectoryStream}. + * If timely disposal of file system resources is required, the + * {@code try}-with-resources construct should be used to ensure that the + * stream's {@link Stream#close close} method is invoked after the stream + * operations are completed. + * + * <p> Operating on a closed stream behaves as if the end of stream * has been reached. Due to read-ahead, one or more elements may be * returned after the stream has been closed. * @@ -3278,7 +3296,7 @@ * * @param dir The path to the directory * - * @return The {@code CloseableStream} describing the content of the + * @return The {@code Stream} describing the content of the * directory * * @throws NotDirectoryException @@ -3294,43 +3312,54 @@ * @see #newDirectoryStream(Path) * @since 1.8 */ - public static CloseableStream<Path> list(Path dir) throws IOException { + public static Stream<Path> list(Path dir) throws IOException { DirectoryStream<Path> ds = Files.newDirectoryStream(dir); - final Iterator<Path> delegate = ds.iterator(); + try { + final Iterator<Path> delegate = ds.iterator(); - // Re-wrap DirectoryIteratorException to UncheckedIOException - Iterator<Path> it = new Iterator<Path>() { - public boolean hasNext() { + // Re-wrap DirectoryIteratorException to UncheckedIOException + Iterator<Path> it = new Iterator<Path>() { + @Override + public boolean hasNext() { + try { + return delegate.hasNext(); + } catch (DirectoryIteratorException e) { + throw new UncheckedIOException(e.getCause()); + } + } + @Override + public Path next() { + try { + return delegate.next(); + } catch (DirectoryIteratorException e) { + throw new UncheckedIOException(e.getCause()); + } + } + }; + + return StreamSupport.stream(Spliterators.spliteratorUnknownSize(it, Spliterator.DISTINCT), false) + .onClose(asUncheckedRunnable(ds)); + } catch (Error|RuntimeException e) { + try { + ds.close(); + } catch (IOException ex) { try { - return delegate.hasNext(); - } catch (DirectoryIteratorException e) { - throw new UncheckedIOException(e.getCause()); - } + e.addSuppressed(ex); + } catch (Throwable ignore) {} } - public Path next() { - try { - return delegate.next(); - } catch (DirectoryIteratorException e) { - throw new UncheckedIOException(e.getCause()); - } - } - }; - - Stream<Path> s = StreamSupport.stream( - Spliterators.spliteratorUnknownSize(it, Spliterator.DISTINCT), - false); - return new DelegatingCloseableStream<>(ds, s); + throw e; + } } /** - * Return a {@code CloseableStream} that is lazily populated with {@code + * Return a {@code Stream} that is lazily populated with {@code * Path} by walking the file tree rooted at a given starting file. The * file tree is traversed <em>depth-first</em>, the elements in the stream * are {@link Path} objects that are obtained as if by {@link * Path#resolve(Path) resolving} the relative path against {@code start}. * * <p> The {@code stream} walks the file tree as elements are consumed. - * The {@code CloseableStream} returned is guaranteed to have at least one + * The {@code Stream} returned is guaranteed to have at least one * element, the starting file itself. For each file visited, the stream * attempts to read its {@link BasicFileAttributes}. If the file is a * directory and can be opened successfully, entries in the directory, and @@ -3370,10 +3399,11 @@ * <p> When a security manager is installed and it denies access to a file * (or directory), then it is ignored and not included in the stream. * - * <p> When not using the try-with-resources construct, then the stream's - * {@link CloseableStream#close close} method should be invoked after the - * operation is completed so as to free any resources held for the open - * directory. Operate the stream after it is closed will throw an + * <p> The returned stream encapsulates one or more {@link DirectoryStream}s. + * If timely disposal of file system resources is required, the + * {@code try}-with-resources construct should be used to ensure that the + * stream's {@link Stream#close close} method is invoked after the stream + * operations are completed. Operating on a closed stream will result in an * {@link java.lang.IllegalStateException}. * * <p> If an {@link IOException} is thrown when accessing the directory @@ -3388,7 +3418,7 @@ * @param options * options to configure the traversal * - * @return the {@link CloseableStream} of {@link Path} + * @return the {@link Stream} of {@link Path} * * @throws IllegalArgumentException * if the {@code maxDepth} parameter is negative @@ -3401,21 +3431,22 @@ * if an I/O error is thrown when accessing the starting file. * @since 1.8 */ - public static CloseableStream<Path> walk(Path start, int maxDepth, - FileVisitOption... options) - throws IOException - { + public static Stream<Path> walk(Path start, int maxDepth, + FileVisitOption... options) + throws IOException { FileTreeIterator iterator = new FileTreeIterator(start, maxDepth, options); - - Stream<Path> s = StreamSupport.stream( - Spliterators.spliteratorUnknownSize(iterator, Spliterator.DISTINCT), - false). - map(entry -> entry.file()); - return new DelegatingCloseableStream<>(iterator, s); + try { + return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.DISTINCT), false) + .onClose(iterator::close) + .map(entry -> entry.file()); + } catch (Error|RuntimeException e) { + iterator.close(); + throw e; + } } /** - * Return a {@code CloseableStream} that is lazily populated with {@code + * Return a {@code Stream} that is lazily populated with {@code * Path} by walking the file tree rooted at a given starting file. The * file tree is traversed <em>depth-first</em>, the elements in the stream * are {@link Path} objects that are obtained as if by {@link @@ -3428,12 +3459,19 @@ * </pre></blockquote> * In other words, it visits all levels of the file tree. * + * <p> The returned stream encapsulates one or more {@link DirectoryStream}s. + * If timely disposal of file system resources is required, the + * {@code try}-with-resources construct should be used to ensure that the + * stream's {@link Stream#close close} method is invoked after the stream + * operations are completed. Operating on a closed stream will result in an + * {@link java.lang.IllegalStateException}. + * * @param start * the starting file * @param options * options to configure the traversal * - * @return the {@link CloseableStream} of {@link Path} + * @return the {@link Stream} of {@link Path} * * @throws SecurityException * If the security manager denies access to the starting file. @@ -3446,15 +3484,14 @@ * @see #walk(Path, int, FileVisitOption...) * @since 1.8 */ - public static CloseableStream<Path> walk(Path start, - FileVisitOption... options) - throws IOException - { + public static Stream<Path> walk(Path start, + FileVisitOption... options) + throws IOException { return walk(start, Integer.MAX_VALUE, options); } /** - * Return a {@code CloseableStream} that is lazily populated with {@code + * Return a {@code Stream} that is lazily populated with {@code * Path} by searching for files in a file tree rooted at a given starting * file. * @@ -3463,12 +3500,19 @@ * {@link BiPredicate} is invoked with its {@link Path} and {@link * BasicFileAttributes}. The {@code Path} object is obtained as if by * {@link Path#resolve(Path) resolving} the relative path against {@code - * start} and is only included in the returned {@link CloseableStream} if + * start} and is only included in the returned {@link Stream} if * the {@code BiPredicate} returns true. Compare to calling {@link * java.util.stream.Stream#filter filter} on the {@code Stream} * returned by {@code walk} method, this method may be more efficient by * avoiding redundant retrieval of the {@code BasicFileAttributes}. * + * <p> The returned stream encapsulates one or more {@link DirectoryStream}s. + * If timely disposal of file system resources is required, the + * {@code try}-with-resources construct should be used to ensure that the + * stream's {@link Stream#close close} method is invoked after the stream + * operations are completed. Operating on a closed stream will result in an + * {@link java.lang.IllegalStateException}. + * * <p> If an {@link IOException} is thrown when accessing the directory * after returned from this method, it is wrapped in an {@link * UncheckedIOException} which will be thrown from the method that caused @@ -3484,7 +3528,7 @@ * @param options * options to configure the traversal * - * @return the {@link CloseableStream} of {@link Path} + * @return the {@link Stream} of {@link Path} * * @throws IllegalArgumentException * if the {@code maxDepth} parameter is negative @@ -3499,24 +3543,25 @@ * @see #walk(Path, int, FileVisitOption...) * @since 1.8 */ - public static CloseableStream<Path> find(Path start, - int maxDepth, - BiPredicate<Path, BasicFileAttributes> matcher, - FileVisitOption... options) - throws IOException - { + public static Stream<Path> find(Path start, + int maxDepth, + BiPredicate<Path, BasicFileAttributes> matcher, + FileVisitOption... options) + throws IOException { FileTreeIterator iterator = new FileTreeIterator(start, maxDepth, options); - - Stream<Path> s = StreamSupport.stream( - Spliterators.spliteratorUnknownSize(iterator, Spliterator.DISTINCT), - false). - filter(entry -> matcher.test(entry.file(), entry.attributes())). - map(entry -> entry.file()); - return new DelegatingCloseableStream<>(iterator, s); + try { + return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.DISTINCT), false) + .onClose(iterator::close) + .filter(entry -> matcher.test(entry.file(), entry.attributes())) + .map(entry -> entry.file()); + } catch (Error|RuntimeException e) { + iterator.close(); + throw e; + } } /** - * Read all lines from a file as a {@code CloseableStream}. Unlike {@link + * Read all lines from a file as a {@code Stream}. Unlike {@link * #readAllLines(Path, Charset) readAllLines}, this method does not read * all lines into a {@code List}, but instead populates lazily as the stream * is consumed. @@ -3528,22 +3573,24 @@ * <p> After this method returns, then any subsequent I/O exception that * occurs while reading from the file or when a malformed or unmappable byte * sequence is read, is wrapped in an {@link UncheckedIOException} that will - * be thrown form the + * be thrown from the * {@link java.util.stream.Stream} method that caused the read to take * place. In case an {@code IOException} is thrown when closing the file, * it is also wrapped as an {@code UncheckedIOException}. * - * <p> When not using the try-with-resources construct, then stream's - * {@link CloseableStream#close close} method should be invoked after - * operation is completed so as to free any resources held for the open - * file. + * <p> The returned stream encapsulates a {@link Reader}. If timely + * disposal of file system resources is required, the try-with-resources + * construct should be used to ensure that the stream's + * {@link Stream#close close} method is invoked after the stream operations + * are completed. + * * * @param path * the path to the file * @param cs * the charset to use for decoding * - * @return the lines from the file as a {@code CloseableStream} + * @return the lines from the file as a {@code Stream} * * @throws IOException * if an I/O error occurs opening the file @@ -3557,10 +3604,19 @@ * @see java.io.BufferedReader#lines() * @since 1.8 */ - public static CloseableStream<String> lines(Path path, Charset cs) - throws IOException - { + public static Stream<String> lines(Path path, Charset cs) throws IOException { BufferedReader br = Files.newBufferedReader(path, cs); - return new DelegatingCloseableStream<>(br, br.lines()); + try { + return br.lines().onClose(asUncheckedRunnable(br)); + } catch (Error|RuntimeException e) { + try { + br.close(); + } catch (IOException ex) { + try { + e.addSuppressed(ex); + } catch (Throwable ignore) {} + } + throw e; + } } }
--- a/src/share/classes/java/util/stream/AbstractPipeline.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/util/stream/AbstractPipeline.java Tue Sep 03 12:16:01 2013 -0700 @@ -71,6 +71,9 @@ */ abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>> extends PipelineHelper<E_OUT> implements BaseStream<E_OUT, S> { + private static final String MSG_STREAM_LINKED = "stream has already been operated upon or closed"; + private static final String MSG_CONSUMED = "source already consumed or closed"; + /** * Backlink to the head of the pipeline chain (self if this is the source * stage). @@ -137,6 +140,8 @@ */ private boolean sourceAnyStateful; + private Runnable sourceCloseAction; + /** * True if pipeline is parallel, otherwise the pipeline is sequential; only * valid for the source stage. @@ -195,7 +200,7 @@ */ AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) { if (previousStage.linkedOrConsumed) - throw new IllegalStateException("stream has already been operated upon"); + throw new IllegalStateException(MSG_STREAM_LINKED); previousStage.linkedOrConsumed = true; previousStage.nextStage = this; @@ -221,7 +226,7 @@ final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) { assert getOutputShape() == terminalOp.inputShape(); if (linkedOrConsumed) - throw new IllegalStateException("stream has already been operated upon"); + throw new IllegalStateException(MSG_STREAM_LINKED); linkedOrConsumed = true; return isParallel() @@ -238,7 +243,7 @@ @SuppressWarnings("unchecked") final Node<E_OUT> evaluateToArrayNode(IntFunction<E_OUT[]> generator) { if (linkedOrConsumed) - throw new IllegalStateException("stream has already been operated upon"); + throw new IllegalStateException(MSG_STREAM_LINKED); linkedOrConsumed = true; // If the last intermediate operation is stateful then @@ -266,7 +271,7 @@ throw new IllegalStateException(); if (linkedOrConsumed) - throw new IllegalStateException("stream has already been operated upon"); + throw new IllegalStateException(MSG_STREAM_LINKED); linkedOrConsumed = true; if (sourceStage.sourceSpliterator != null) { @@ -282,7 +287,7 @@ return s; } else { - throw new IllegalStateException("source already consumed"); + throw new IllegalStateException(MSG_CONSUMED); } } @@ -302,12 +307,35 @@ return (S) this; } + @Override + public void close() { + linkedOrConsumed = true; + sourceSupplier = null; + sourceSpliterator = null; + if (sourceStage.sourceCloseAction != null) { + Runnable closeAction = sourceStage.sourceCloseAction; + sourceStage.sourceCloseAction = null; + closeAction.run(); + } + } + + @Override + @SuppressWarnings("unchecked") + public S onClose(Runnable closeHandler) { + Runnable existingHandler = sourceStage.sourceCloseAction; + sourceStage.sourceCloseAction = + (existingHandler == null) + ? closeHandler + : Streams.composeWithExceptions(existingHandler, closeHandler); + return (S) this; + } + // Primitive specialization use co-variant overrides, hence is not final @Override @SuppressWarnings("unchecked") public Spliterator<E_OUT> spliterator() { if (linkedOrConsumed) - throw new IllegalStateException("stream has already been operated upon"); + throw new IllegalStateException(MSG_STREAM_LINKED); linkedOrConsumed = true; if (this == sourceStage) { @@ -324,7 +352,7 @@ return lazySpliterator(s); } else { - throw new IllegalStateException("source already consumed"); + throw new IllegalStateException(MSG_CONSUMED); } } else { @@ -424,7 +452,7 @@ sourceStage.sourceSupplier = null; } else { - throw new IllegalStateException("source already consumed"); + throw new IllegalStateException(MSG_CONSUMED); } if (isParallel()) {
--- a/src/share/classes/java/util/stream/BaseStream.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/util/stream/BaseStream.java Tue Sep 03 12:16:01 2013 -0700 @@ -35,7 +35,8 @@ * @param <S> type of stream implementing {@code BaseStream} * @since 1.8 */ -public interface BaseStream<T, S extends BaseStream<T, S>> { +public interface BaseStream<T, S extends BaseStream<T, S>> + extends AutoCloseable { /** * Returns an iterator for the elements of this stream. * @@ -103,4 +104,33 @@ * @return an unordered stream */ S unordered(); + + /** + * Returns an equivalent stream with an additional close handler. Close + * handlers are run when the {@link #close()} method + * is called on the stream, and are executed in the order they were + * added. All close handlers are run, even if earlier close handlers throw + * exceptions. If any close handler throws an exception, the first + * exception thrown will be relayed to the caller of {@code close()}, with + * any remaining exceptions added to that exception as suppressed exceptions + * (unless one of the remaining exceptions is the same exception as the + * first exception, since an exception cannot suppress itself.) May + * return itself. + * + * <p>This is an <a href="package-summary.html#StreamOps">intermediate + * operation</a>. + * + * @param closeHandler A task to execute when the stream is closed + * @return a stream with a handler that is run if the stream is closed + */ + S onClose(Runnable closeHandler); + + /** + * Closes this stream, causing all close handlers for this stream pipeline + * to be called. + * + * @see AutoCloseable#close() + */ + @Override + void close(); }
--- a/src/share/classes/java/util/stream/CloseableStream.java Tue Sep 03 11:44:34 2013 -0700 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,57 +0,0 @@ -/* - * Copyright (c) 2013, Oracle and/or its affiliates. All rights reserved. - * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. - * - * This code is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License version 2 only, as - * published by the Free Software Foundation. Oracle designates this - * particular file as subject to the "Classpath" exception as provided - * by Oracle in the LICENSE file that accompanied this code. - * - * This code is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or - * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License - * version 2 for more details (a copy is included in the LICENSE file that - * accompanied this code). - * - * You should have received a copy of the GNU General Public License version - * 2 along with this work; if not, write to the Free Software Foundation, - * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. - * - * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA - * or visit www.oracle.com if you need additional information or have any - * questions. - */ - -package java.util.stream; - -/** - * A {@code CloseableStream} is a {@code Stream} that can be closed. - * The close method is invoked to release resources that the object is - * holding (such as open files). - * - * @param <T> The type of stream elements - * @since 1.8 - */ -public interface CloseableStream<T> extends Stream<T>, AutoCloseable { - - /** - * Closes this resource, relinquishing any underlying resources. - * This method is invoked automatically on objects managed by the - * {@code try}-with-resources statement. Does nothing if called when - * the resource has already been closed. - * - * This method does not allow throwing checked {@code Exception}s like - * {@link AutoCloseable#close() AutoCloseable.close()}. Cases where the - * close operation may fail require careful attention by implementers. It - * is strongly advised to relinquish the underlying resources and to - * internally <em>mark</em> the resource as closed. The {@code close} - * method is unlikely to be invoked more than once and so this ensures - * that the resources are released in a timely manner. Furthermore it - * reduces problems that could arise when the resource wraps, or is - * wrapped, by another resource. - * - * @see AutoCloseable#close() - */ - void close(); -}
--- a/src/share/classes/java/util/stream/DelegatingStream.java Tue Sep 03 11:44:34 2013 -0700 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,270 +0,0 @@ -/* - * Copyright (c) 2013, Oracle and/or its affiliates. All rights reserved. - * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. - * - * This code is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License version 2 only, as - * published by the Free Software Foundation. Oracle designates this - * particular file as subject to the "Classpath" exception as provided - * by Oracle in the LICENSE file that accompanied this code. - * - * This code is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or - * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License - * version 2 for more details (a copy is included in the LICENSE file that - * accompanied this code). - * - * You should have received a copy of the GNU General Public License version - * 2 along with this work; if not, write to the Free Software Foundation, - * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. - * - * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA - * or visit www.oracle.com if you need additional information or have any - * questions. - */ - -package java.util.stream; - -import java.util.Comparator; -import java.util.Iterator; -import java.util.Objects; -import java.util.Optional; -import java.util.Spliterator; -import java.util.function.BiConsumer; -import java.util.function.BiFunction; -import java.util.function.BinaryOperator; -import java.util.function.Consumer; -import java.util.function.Function; -import java.util.function.IntFunction; -import java.util.function.Predicate; -import java.util.function.Supplier; -import java.util.function.ToDoubleFunction; -import java.util.function.ToIntFunction; -import java.util.function.ToLongFunction; - -/** - * A {@code Stream} implementation that delegates operations to another {@code - * Stream}. - * - * @param <T> type of stream elements for this stream and underlying delegate - * stream - * - * @since 1.8 - */ -public class DelegatingStream<T> implements Stream<T> { - final private Stream<T> delegate; - - /** - * Construct a {@code Stream} that delegates operations to another {@code - * Stream}. - * - * @param delegate the underlying {@link Stream} to which we delegate all - * {@code Stream} methods - * @throws NullPointerException if the delegate is null - */ - public DelegatingStream(Stream<T> delegate) { - this.delegate = Objects.requireNonNull(delegate); - } - - // -- BaseStream methods -- - - @Override - public Spliterator<T> spliterator() { - return delegate.spliterator(); - } - - @Override - public boolean isParallel() { - return delegate.isParallel(); - } - - @Override - public Iterator<T> iterator() { - return delegate.iterator(); - } - - // -- Stream methods -- - - @Override - public Stream<T> filter(Predicate<? super T> predicate) { - return delegate.filter(predicate); - } - - @Override - public <R> Stream<R> map(Function<? super T, ? extends R> mapper) { - return delegate.map(mapper); - } - - @Override - public IntStream mapToInt(ToIntFunction<? super T> mapper) { - return delegate.mapToInt(mapper); - } - - @Override - public LongStream mapToLong(ToLongFunction<? super T> mapper) { - return delegate.mapToLong(mapper); - } - - @Override - public DoubleStream mapToDouble(ToDoubleFunction<? super T> mapper) { - return delegate.mapToDouble(mapper); - } - - @Override - public <R> Stream<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper) { - return delegate.flatMap(mapper); - } - - @Override - public IntStream flatMapToInt(Function<? super T, ? extends IntStream> mapper) { - return delegate.flatMapToInt(mapper); - } - - @Override - public LongStream flatMapToLong(Function<? super T, ? extends LongStream> mapper) { - return delegate.flatMapToLong(mapper); - } - - @Override - public DoubleStream flatMapToDouble(Function<? super T, ? extends DoubleStream> mapper) { - return delegate.flatMapToDouble(mapper); - } - - @Override - public Stream<T> distinct() { - return delegate.distinct(); - } - - @Override - public Stream<T> sorted() { - return delegate.sorted(); - } - - @Override - public Stream<T> sorted(Comparator<? super T> comparator) { - return delegate.sorted(comparator); - } - - @Override - public void forEach(Consumer<? super T> action) { - delegate.forEach(action); - } - - @Override - public void forEachOrdered(Consumer<? super T> action) { - delegate.forEachOrdered(action); - } - - @Override - public Stream<T> peek(Consumer<? super T> consumer) { - return delegate.peek(consumer); - } - - @Override - public Stream<T> limit(long maxSize) { - return delegate.limit(maxSize); - } - - @Override - public Stream<T> substream(long startingOffset) { - return delegate.substream(startingOffset); - } - - @Override - public Stream<T> substream(long startingOffset, long endingOffset) { - return delegate.substream(startingOffset, endingOffset); - } - - @Override - public <A> A[] toArray(IntFunction<A[]> generator) { - return delegate.toArray(generator); - } - - @Override - public Object[] toArray() { - return delegate.toArray(); - } - - @Override - public T reduce(T identity, BinaryOperator<T> accumulator) { - return delegate.reduce(identity, accumulator); - } - - @Override - public Optional<T> reduce(BinaryOperator<T> accumulator) { - return delegate.reduce(accumulator); - } - - @Override - public <U> U reduce(U identity, BiFunction<U, ? super T, U> accumulator, - BinaryOperator<U> combiner) { - return delegate.reduce(identity, accumulator, combiner); - } - - @Override - public <R> R collect(Supplier<R> resultFactory, - BiConsumer<R, ? super T> accumulator, - BiConsumer<R, R> combiner) { - return delegate.collect(resultFactory, accumulator, combiner); - } - - @Override - public <R, A> R collect(Collector<? super T, A, R> collector) { - return delegate.collect(collector); - } - - @Override - public Optional<T> max(Comparator<? super T> comparator) { - return delegate.max(comparator); - } - - @Override - public Optional<T> min(Comparator<? super T> comparator) { - return delegate.min(comparator); - } - - @Override - public long count() { - return delegate.count(); - } - - @Override - public boolean anyMatch(Predicate<? super T> predicate) { - return delegate.anyMatch(predicate); - } - - @Override - public boolean allMatch(Predicate<? super T> predicate) { - return delegate.allMatch(predicate); - } - - @Override - public boolean noneMatch(Predicate<? super T> predicate) { - return delegate.noneMatch(predicate); - } - - @Override - public Optional<T> findFirst() { - return delegate.findFirst(); - } - - @Override - public Optional<T> findAny() { - return delegate.findAny(); - } - - @Override - public Stream<T> unordered() { - return delegate.unordered(); - } - - @Override - public Stream<T> sequential() { - return delegate.sequential(); - } - - @Override - public Stream<T> parallel() { - return delegate.parallel(); - } -}
--- a/src/share/classes/java/util/stream/DoublePipeline.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/util/stream/DoublePipeline.java Tue Sep 03 12:16:01 2013 -0700 @@ -266,10 +266,11 @@ @Override public void accept(double t) { - // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it - DoubleStream result = mapper.apply(t); - if (result != null) - result.sequential().forEach(i -> downstream.accept(i)); + try (DoubleStream result = mapper.apply(t)) { + // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it + if (result != null) + result.sequential().forEach(i -> downstream.accept(i)); + } } }; }
--- a/src/share/classes/java/util/stream/DoubleStream.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/util/stream/DoubleStream.java Tue Sep 03 12:16:01 2013 -0700 @@ -752,7 +752,8 @@ * elements of a first {@code DoubleStream} succeeded by all the elements of the * second {@code DoubleStream}. The resulting stream is ordered if both * of the input streams are ordered, and parallel if either of the input - * streams is parallel. + * streams is parallel. When the resulting stream is closed, the close + * handlers for both input streams are invoked. * * @param a the first stream * @param b the second stream to concatenate on to end of the first stream @@ -764,7 +765,8 @@ Spliterator.OfDouble split = new Streams.ConcatSpliterator.OfDouble( a.spliterator(), b.spliterator()); - return StreamSupport.doubleStream(split, a.isParallel() || b.isParallel()); + DoubleStream stream = StreamSupport.doubleStream(split, a.isParallel() || b.isParallel()); + return stream.onClose(Streams.composedClose(a, b)); } /**
--- a/src/share/classes/java/util/stream/IntPipeline.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/util/stream/IntPipeline.java Tue Sep 03 12:16:01 2013 -0700 @@ -302,10 +302,11 @@ @Override public void accept(int t) { - // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it - IntStream result = mapper.apply(t); - if (result != null) - result.sequential().forEach(i -> downstream.accept(i)); + try (IntStream result = mapper.apply(t)) { + // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it + if (result != null) + result.sequential().forEach(i -> downstream.accept(i)); + } } }; }
--- a/src/share/classes/java/util/stream/IntStream.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/util/stream/IntStream.java Tue Sep 03 12:16:01 2013 -0700 @@ -806,7 +806,8 @@ * elements of a first {@code IntStream} succeeded by all the elements of the * second {@code IntStream}. The resulting stream is ordered if both * of the input streams are ordered, and parallel if either of the input - * streams is parallel. + * streams is parallel. When the resulting stream is closed, the close + * handlers for both input streams are invoked. * * @param a the first stream * @param b the second stream to concatenate on to end of the first stream @@ -818,7 +819,8 @@ Spliterator.OfInt split = new Streams.ConcatSpliterator.OfInt( a.spliterator(), b.spliterator()); - return StreamSupport.intStream(split, a.isParallel() || b.isParallel()); + IntStream stream = StreamSupport.intStream(split, a.isParallel() || b.isParallel()); + return stream.onClose(Streams.composedClose(a, b)); } /**
--- a/src/share/classes/java/util/stream/LongPipeline.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/util/stream/LongPipeline.java Tue Sep 03 12:16:01 2013 -0700 @@ -283,10 +283,11 @@ @Override public void accept(long t) { - // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it - LongStream result = mapper.apply(t); - if (result != null) - result.sequential().forEach(i -> downstream.accept(i)); + try (LongStream result = mapper.apply(t)) { + // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it + if (result != null) + result.sequential().forEach(i -> downstream.accept(i)); + } } }; }
--- a/src/share/classes/java/util/stream/LongStream.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/util/stream/LongStream.java Tue Sep 03 12:16:01 2013 -0700 @@ -812,7 +812,8 @@ * elements of a first {@code LongStream} succeeded by all the elements of the * second {@code LongStream}. The resulting stream is ordered if both * of the input streams are ordered, and parallel if either of the input - * streams is parallel. + * streams is parallel. When the resulting stream is closed, the close + * handlers for both input streams are invoked. * * @param a the first stream * @param b the second stream to concatenate on to end of the first stream @@ -824,7 +825,8 @@ Spliterator.OfLong split = new Streams.ConcatSpliterator.OfLong( a.spliterator(), b.spliterator()); - return StreamSupport.longStream(split, a.isParallel() || b.isParallel()); + LongStream stream = StreamSupport.longStream(split, a.isParallel() || b.isParallel()); + return stream.onClose(Streams.composedClose(a, b)); } /**
--- a/src/share/classes/java/util/stream/ReferencePipeline.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/util/stream/ReferencePipeline.java Tue Sep 03 12:16:01 2013 -0700 @@ -264,10 +264,11 @@ @Override public void accept(P_OUT u) { - // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it - Stream<? extends R> result = mapper.apply(u); - if (result != null) - result.sequential().forEach(downstream); + try (Stream<? extends R> result = mapper.apply(u)) { + // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it + if (result != null) + result.sequential().forEach(downstream); + } } }; } @@ -291,10 +292,11 @@ @Override public void accept(P_OUT u) { - // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it - IntStream result = mapper.apply(u); - if (result != null) - result.sequential().forEach(downstreamAsInt); + try (IntStream result = mapper.apply(u)) { + // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it + if (result != null) + result.sequential().forEach(downstreamAsInt); + } } }; } @@ -318,10 +320,11 @@ @Override public void accept(P_OUT u) { - // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it - DoubleStream result = mapper.apply(u); - if (result != null) - result.sequential().forEach(downstreamAsDouble); + try (DoubleStream result = mapper.apply(u)) { + // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it + if (result != null) + result.sequential().forEach(downstreamAsDouble); + } } }; } @@ -345,10 +348,11 @@ @Override public void accept(P_OUT u) { - // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it - LongStream result = mapper.apply(u); - if (result != null) - result.sequential().forEach(downstreamAsLong); + try (LongStream result = mapper.apply(u)) { + // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it + if (result != null) + result.sequential().forEach(downstreamAsLong); + } } }; }
--- a/src/share/classes/java/util/stream/Stream.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/util/stream/Stream.java Tue Sep 03 12:16:01 2013 -0700 @@ -891,7 +891,8 @@ * elements of a first {@code Stream} succeeded by all the elements of the * second {@code Stream}. The resulting stream is ordered if both * of the input streams are ordered, and parallel if either of the input - * streams is parallel. + * streams is parallel. When the resulting stream is closed, the close + * handlers for both input streams are invoked. * * @param <T> The type of stream elements * @param a the first stream @@ -906,7 +907,8 @@ @SuppressWarnings("unchecked") Spliterator<T> split = new Streams.ConcatSpliterator.OfRef<>( (Spliterator<T>) a.spliterator(), (Spliterator<T>) b.spliterator()); - return StreamSupport.stream(split, a.isParallel() || b.isParallel()); + Stream<T> stream = StreamSupport.stream(split, a.isParallel() || b.isParallel()); + return stream.onClose(Streams.composedClose(a, b)); } /**
--- a/src/share/classes/java/util/stream/Streams.java Tue Sep 03 11:44:34 2013 -0700 +++ b/src/share/classes/java/util/stream/Streams.java Tue Sep 03 12:16:01 2013 -0700 @@ -833,4 +833,61 @@ } } } + + /** + * Given two Runnables, return a Runnable that executes both in sequence, + * even if the first throws an exception, and if both throw exceptions, add + * any exceptions thrown by the second as suppressed exceptions of the first. + */ + static Runnable composeWithExceptions(Runnable a, Runnable b) { + return new Runnable() { + @Override + public void run() { + try { + a.run(); + } + catch (Throwable e1) { + try { + b.run(); + } + catch (Throwable e2) { + try { + e1.addSuppressed(e2); + } catch (Throwable ignore) {} + } + throw e1; + } + b.run(); + } + }; + } + + /** + * Given two streams, return a Runnable that + * executes both of their {@link BaseStream#close} methods in sequence, + * even if the first throws an exception, and if both throw exceptions, add + * any exceptions thrown by the second as suppressed exceptions of the first. + */ + static Runnable composedClose(BaseStream<?, ?> a, BaseStream<?, ?> b) { + return new Runnable() { + @Override + public void run() { + try { + a.close(); + } + catch (Throwable e1) { + try { + b.close(); + } + catch (Throwable e2) { + try { + e1.addSuppressed(e2); + } catch (Throwable ignore) {} + } + throw e1; + } + b.close(); + } + }; + } }
--- a/test/java/nio/file/Files/StreamTest.java Tue Sep 03 11:44:34 2013 -0700 +++ b/test/java/nio/file/Files/StreamTest.java Tue Sep 03 12:16:01 2013 -0700 @@ -43,14 +43,13 @@ import java.nio.file.Paths; import java.nio.file.attribute.BasicFileAttributes; import java.util.Arrays; -import java.util.Comparator; import java.util.Iterator; import java.util.List; import java.util.Objects; import java.util.Set; import java.util.TreeSet; import java.util.function.BiPredicate; -import java.util.stream.CloseableStream; +import java.util.stream.Stream; import java.util.stream.Collectors; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; @@ -138,14 +137,14 @@ } public void testBasic() { - try (CloseableStream<Path> s = Files.list(testFolder)) { - Object[] actual = s.sorted(Comparator.naturalOrder()).toArray(); + try (Stream<Path> s = Files.list(testFolder)) { + Object[] actual = s.sorted().toArray(); assertEquals(actual, level1); } catch (IOException ioe) { fail("Unexpected IOException"); } - try (CloseableStream<Path> s = Files.list(testFolder.resolve("empty"))) { + try (Stream<Path> s = Files.list(testFolder.resolve("empty"))) { int count = s.mapToInt(p -> 1).reduce(0, Integer::sum); assertEquals(count, 0, "Expect empty stream."); } catch (IOException ioe) { @@ -154,8 +153,8 @@ } public void testWalk() { - try (CloseableStream<Path> s = Files.walk(testFolder)) { - Object[] actual = s.sorted(Comparator.naturalOrder()).toArray(); + try (Stream<Path> s = Files.walk(testFolder)) { + Object[] actual = s.sorted().toArray(); assertEquals(actual, all); } catch (IOException ioe) { fail("Unexpected IOException"); @@ -163,9 +162,9 @@ } public void testWalkOneLevel() { - try (CloseableStream<Path> s = Files.walk(testFolder, 1)) { + try (Stream<Path> s = Files.walk(testFolder, 1)) { Object[] actual = s.filter(path -> ! path.equals(testFolder)) - .sorted(Comparator.naturalOrder()) + .sorted() .toArray(); assertEquals(actual, level1); } catch (IOException ioe) { @@ -176,8 +175,8 @@ public void testWalkFollowLink() { // If link is not supported, the directory structure won't have link. // We still want to test the behavior with FOLLOW_LINKS option. - try (CloseableStream<Path> s = Files.walk(testFolder, FileVisitOption.FOLLOW_LINKS)) { - Object[] actual = s.sorted(Comparator.naturalOrder()).toArray(); + try (Stream<Path> s = Files.walk(testFolder, FileVisitOption.FOLLOW_LINKS)) { + Object[] actual = s.sorted().toArray(); assertEquals(actual, all_folowLinks); } catch (IOException ioe) { fail("Unexpected IOException"); @@ -185,7 +184,7 @@ } private void validateFileSystemLoopException(Path start, Path... causes) { - try (CloseableStream<Path> s = Files.walk(start, FileVisitOption.FOLLOW_LINKS)) { + try (Stream<Path> s = Files.walk(start, FileVisitOption.FOLLOW_LINKS)) { try { int count = s.mapToInt(p -> 1).reduce(0, Integer::sum); fail("Should got FileSystemLoopException, but got " + count + "elements."); @@ -282,28 +281,28 @@ public void testFind() throws IOException { PathBiPredicate pred = new PathBiPredicate((path, attrs) -> true); - try (CloseableStream<Path> s = Files.find(testFolder, Integer.MAX_VALUE, pred)) { + try (Stream<Path> s = Files.find(testFolder, Integer.MAX_VALUE, pred)) { Set<Path> result = s.collect(Collectors.toCollection(TreeSet::new)); assertEquals(pred.visited(), all); assertEquals(result.toArray(new Path[0]), pred.visited()); } pred = new PathBiPredicate((path, attrs) -> attrs.isSymbolicLink()); - try (CloseableStream<Path> s = Files.find(testFolder, Integer.MAX_VALUE, pred)) { + try (Stream<Path> s = Files.find(testFolder, Integer.MAX_VALUE, pred)) { s.forEach(path -> assertTrue(Files.isSymbolicLink(path))); assertEquals(pred.visited(), all); } pred = new PathBiPredicate((path, attrs) -> path.getFileName().toString().startsWith("e")); - try (CloseableStream<Path> s = Files.find(testFolder, Integer.MAX_VALUE, pred)) { + try (Stream<Path> s = Files.find(testFolder, Integer.MAX_VALUE, pred)) { s.forEach(path -> assertEquals(path.getFileName().toString(), "empty")); assertEquals(pred.visited(), all); } pred = new PathBiPredicate((path, attrs) -> path.getFileName().toString().startsWith("l") && attrs.isRegularFile()); - try (CloseableStream<Path> s = Files.find(testFolder, Integer.MAX_VALUE, pred)) { + try (Stream<Path> s = Files.find(testFolder, Integer.MAX_VALUE, pred)) { s.forEach(path -> fail("Expect empty stream")); assertEquals(pred.visited(), all); } @@ -317,14 +316,14 @@ try { // zero lines assertTrue(Files.size(tmpfile) == 0, "File should be empty"); - try (CloseableStream<String> s = Files.lines(tmpfile, US_ASCII)) { + try (Stream<String> s = Files.lines(tmpfile, US_ASCII)) { assertEquals(s.mapToInt(l -> 1).reduce(0, Integer::sum), 0, "No line expected"); } // one line byte[] hi = { (byte)'h', (byte)'i' }; Files.write(tmpfile, hi); - try (CloseableStream<String> s = Files.lines(tmpfile, US_ASCII)) { + try (Stream<String> s = Files.lines(tmpfile, US_ASCII)) { List<String> lines = s.collect(Collectors.toList()); assertTrue(lines.size() == 1, "One line expected"); assertTrue(lines.get(0).equals("hi"), "'Hi' expected"); @@ -334,7 +333,7 @@ List<String> expected = Arrays.asList("hi", "there"); Files.write(tmpfile, expected, US_ASCII); assertTrue(Files.size(tmpfile) > 0, "File is empty"); - try (CloseableStream<String> s = Files.lines(tmpfile, US_ASCII)) { + try (Stream<String> s = Files.lines(tmpfile, US_ASCII)) { List<String> lines = s.collect(Collectors.toList()); assertTrue(lines.equals(expected), "Unexpected lines"); } @@ -342,7 +341,7 @@ // MalformedInputException byte[] bad = { (byte)0xff, (byte)0xff }; Files.write(tmpfile, bad); - try (CloseableStream<String> s = Files.lines(tmpfile, US_ASCII)) { + try (Stream<String> s = Files.lines(tmpfile, US_ASCII)) { try { List<String> lines = s.collect(Collectors.toList()); throw new RuntimeException("UncheckedIOException expected"); @@ -378,7 +377,7 @@ fsp.setFaultyMode(false); Path fakeRoot = fs.getRoot(); try { - try (CloseableStream<Path> s = Files.list(fakeRoot)) { + try (Stream<Path> s = Files.list(fakeRoot)) { s.forEach(path -> assertEquals(path.getFileName().toString(), "DirectoryIteratorException")); } } catch (UncheckedIOException uioe) { @@ -398,7 +397,7 @@ } try { - try (CloseableStream<Path> s = Files.list(fakeRoot)) { + try (Stream<Path> s = Files.list(fakeRoot)) { s.forEach(path -> fail("should not get here")); } } catch (UncheckedIOException uioe) { @@ -427,12 +426,12 @@ try { fsp.setFaultyMode(false); Path fakeRoot = fs.getRoot(); - try (CloseableStream<Path> s = Files.list(fakeRoot.resolve("dir2"))) { + try (Stream<Path> s = Files.list(fakeRoot.resolve("dir2"))) { // only one file s.forEach(path -> assertEquals(path.getFileName().toString(), "IOException")); } - try (CloseableStream<Path> s = Files.walk(fakeRoot.resolve("empty"))) { + try (Stream<Path> s = Files.walk(fakeRoot.resolve("empty"))) { String[] result = s.map(path -> path.getFileName().toString()) .toArray(String[]::new); // ordered as depth-first @@ -440,13 +439,13 @@ } fsp.setFaultyMode(true); - try (CloseableStream<Path> s = Files.list(fakeRoot.resolve("dir2"))) { + try (Stream<Path> s = Files.list(fakeRoot.resolve("dir2"))) { s.forEach(path -> fail("should have caused exception")); } catch (UncheckedIOException uioe) { assertTrue(uioe.getCause() instanceof FaultyFileSystem.FaultyException); } - try (CloseableStream<Path> s = Files.walk(fakeRoot.resolve("empty"))) { + try (Stream<Path> s = Files.walk(fakeRoot.resolve("empty"))) { String[] result = s.map(path -> path.getFileName().toString()) .toArray(String[]::new); fail("should not reach here due to IOException"); @@ -454,7 +453,7 @@ assertTrue(uioe.getCause() instanceof FaultyFileSystem.FaultyException); } - try (CloseableStream<Path> s = Files.walk( + try (Stream<Path> s = Files.walk( fakeRoot.resolve("empty").resolve("IOException"))) { String[] result = s.map(path -> path.getFileName().toString()) @@ -502,20 +501,20 @@ fsp.setFaultyMode(false); Path fakeRoot = fs.getRoot(); // validate setting - try (CloseableStream<Path> s = Files.list(fakeRoot.resolve("empty"))) { + try (Stream<Path> s = Files.list(fakeRoot.resolve("empty"))) { String[] result = s.map(path -> path.getFileName().toString()) .toArray(String[]::new); assertEqualsNoOrder(result, new String[] { "SecurityException", "sample" }); } - try (CloseableStream<Path> s = Files.walk(fakeRoot.resolve("dir2"))) { + try (Stream<Path> s = Files.walk(fakeRoot.resolve("dir2"))) { String[] result = s.map(path -> path.getFileName().toString()) .toArray(String[]::new); assertEqualsNoOrder(result, new String[] { "dir2", "SecurityException", "fileInSE", "file" }); } if (supportsLinks) { - try (CloseableStream<Path> s = Files.list(fakeRoot.resolve("dir"))) { + try (Stream<Path> s = Files.list(fakeRoot.resolve("dir"))) { String[] result = s.map(path -> path.getFileName().toString()) .toArray(String[]::new); assertEqualsNoOrder(result, new String[] { "d1", "f1", "lnDir2", "SecurityException", "lnDirSE", "lnFileSE" }); @@ -525,13 +524,13 @@ // execute test fsp.setFaultyMode(true); // ignore file cause SecurityException - try (CloseableStream<Path> s = Files.walk(fakeRoot.resolve("empty"))) { + try (Stream<Path> s = Files.walk(fakeRoot.resolve("empty"))) { String[] result = s.map(path -> path.getFileName().toString()) .toArray(String[]::new); assertEqualsNoOrder(result, new String[] { "empty", "sample" }); } // skip folder cause SecurityException - try (CloseableStream<Path> s = Files.walk(fakeRoot.resolve("dir2"))) { + try (Stream<Path> s = Files.walk(fakeRoot.resolve("dir2"))) { String[] result = s.map(path -> path.getFileName().toString()) .toArray(String[]::new); assertEqualsNoOrder(result, new String[] { "dir2", "file" }); @@ -539,14 +538,14 @@ if (supportsLinks) { // not following links - try (CloseableStream<Path> s = Files.walk(fakeRoot.resolve("dir"))) { + try (Stream<Path> s = Files.walk(fakeRoot.resolve("dir"))) { String[] result = s.map(path -> path.getFileName().toString()) .toArray(String[]::new); assertEqualsNoOrder(result, new String[] { "dir", "d1", "f1", "lnDir2", "lnDirSE", "lnFileSE" }); } // following links - try (CloseableStream<Path> s = Files.walk(fakeRoot.resolve("dir"), FileVisitOption.FOLLOW_LINKS)) { + try (Stream<Path> s = Files.walk(fakeRoot.resolve("dir"), FileVisitOption.FOLLOW_LINKS)) { String[] result = s.map(path -> path.getFileName().toString()) .toArray(String[]::new); // ?? Should fileInSE show up? @@ -556,19 +555,19 @@ } // list instead of walk - try (CloseableStream<Path> s = Files.list(fakeRoot.resolve("empty"))) { + try (Stream<Path> s = Files.list(fakeRoot.resolve("empty"))) { String[] result = s.map(path -> path.getFileName().toString()) .toArray(String[]::new); assertEqualsNoOrder(result, new String[] { "sample" }); } - try (CloseableStream<Path> s = Files.list(fakeRoot.resolve("dir2"))) { + try (Stream<Path> s = Files.list(fakeRoot.resolve("dir2"))) { String[] result = s.map(path -> path.getFileName().toString()) .toArray(String[]::new); assertEqualsNoOrder(result, new String[] { "file" }); } // root cause SecurityException should be reported - try (CloseableStream<Path> s = Files.walk( + try (Stream<Path> s = Files.walk( fakeRoot.resolve("dir2").resolve("SecurityException"))) { String[] result = s.map(path -> path.getFileName().toString()) @@ -579,7 +578,7 @@ } // Walk a file cause SecurityException, we should get SE - try (CloseableStream<Path> s = Files.walk( + try (Stream<Path> s = Files.walk( fakeRoot.resolve("dir").resolve("SecurityException"))) { String[] result = s.map(path -> path.getFileName().toString()) @@ -590,7 +589,7 @@ } // List a file cause SecurityException, we should get SE as cannot read attribute - try (CloseableStream<Path> s = Files.list( + try (Stream<Path> s = Files.list( fakeRoot.resolve("dir2").resolve("SecurityException"))) { String[] result = s.map(path -> path.getFileName().toString()) @@ -600,7 +599,7 @@ assertTrue(se.getCause() instanceof FaultyFileSystem.FaultyException); } - try (CloseableStream<Path> s = Files.list( + try (Stream<Path> s = Files.list( fakeRoot.resolve("dir").resolve("SecurityException"))) { String[] result = s.map(path -> path.getFileName().toString()) @@ -627,7 +626,7 @@ } public void testConstructException() { - try (CloseableStream<String> s = Files.lines(testFolder.resolve("notExist"), Charset.forName("UTF-8"))) { + try (Stream<String> s = Files.lines(testFolder.resolve("notExist"), Charset.forName("UTF-8"))) { s.forEach(l -> fail("File is not even exist!")); } catch (IOException ioe) { assertTrue(ioe instanceof NoSuchFileException); @@ -635,24 +634,26 @@ } public void testClosedStream() throws IOException { - try (CloseableStream<Path> s = Files.list(testFolder)) { + try (Stream<Path> s = Files.list(testFolder)) { s.close(); - Object[] actual = s.sorted(Comparator.naturalOrder()).toArray(); - assertTrue(actual.length <= level1.length); - } - - try (CloseableStream<Path> s = Files.walk(testFolder)) { - s.close(); - Object[] actual = s.sorted(Comparator.naturalOrder()).toArray(); + Object[] actual = s.sorted().toArray(); fail("Operate on closed stream should throw IllegalStateException"); } catch (IllegalStateException ex) { // expected } - try (CloseableStream<Path> s = Files.find(testFolder, Integer.MAX_VALUE, + try (Stream<Path> s = Files.walk(testFolder)) { + s.close(); + Object[] actual = s.sorted().toArray(); + fail("Operate on closed stream should throw IllegalStateException"); + } catch (IllegalStateException ex) { + // expected + } + + try (Stream<Path> s = Files.find(testFolder, Integer.MAX_VALUE, (p, attr) -> true)) { s.close(); - Object[] actual = s.sorted(Comparator.naturalOrder()).toArray(); + Object[] actual = s.sorted().toArray(); fail("Operate on closed stream should throw IllegalStateException"); } catch (IllegalStateException ex) { // expected
--- a/test/java/util/stream/bootlib/java/util/stream/DoubleStreamTestScenario.java Tue Sep 03 11:44:34 2013 -0700 +++ b/test/java/util/stream/bootlib/java/util/stream/DoubleStreamTestScenario.java Tue Sep 03 12:16:01 2013 -0700 @@ -40,7 +40,7 @@ @SuppressWarnings({"rawtypes", "unchecked"}) public enum DoubleStreamTestScenario implements OpTestCase.BaseStreamTestScenario { - STREAM_FOR_EACH(false) { + STREAM_FOR_EACH_WITH_CLOSE(false) { <T, S_IN extends BaseStream<T, S_IN>> void _run(TestData<T, S_IN> data, DoubleConsumer b, Function<S_IN, DoubleStream> m) { DoubleStream s = m.apply(data.stream()); @@ -48,6 +48,7 @@ s = s.sequential(); } s.forEach(b); + s.close(); } },
--- a/test/java/util/stream/bootlib/java/util/stream/IntStreamTestScenario.java Tue Sep 03 11:44:34 2013 -0700 +++ b/test/java/util/stream/bootlib/java/util/stream/IntStreamTestScenario.java Tue Sep 03 12:16:01 2013 -0700 @@ -40,7 +40,7 @@ @SuppressWarnings({"rawtypes", "unchecked"}) public enum IntStreamTestScenario implements OpTestCase.BaseStreamTestScenario { - STREAM_FOR_EACH(false) { + STREAM_FOR_EACH_WITH_CLOSE(false) { <T, S_IN extends BaseStream<T, S_IN>> void _run(TestData<T, S_IN> data, IntConsumer b, Function<S_IN, IntStream> m) { IntStream s = m.apply(data.stream()); @@ -48,6 +48,7 @@ s = s.sequential(); } s.forEach(b); + s.close(); } },
--- a/test/java/util/stream/bootlib/java/util/stream/LongStreamTestScenario.java Tue Sep 03 11:44:34 2013 -0700 +++ b/test/java/util/stream/bootlib/java/util/stream/LongStreamTestScenario.java Tue Sep 03 12:16:01 2013 -0700 @@ -40,7 +40,7 @@ @SuppressWarnings({"rawtypes", "unchecked"}) public enum LongStreamTestScenario implements OpTestCase.BaseStreamTestScenario { - STREAM_FOR_EACH(false) { + STREAM_FOR_EACH_WITH_CLOSE(false) { <T, S_IN extends BaseStream<T, S_IN>> void _run(TestData<T, S_IN> data, LongConsumer b, Function<S_IN, LongStream> m) { LongStream s = m.apply(data.stream()); @@ -48,6 +48,7 @@ s = s.sequential(); } s.forEach(b); + s.close(); } },
--- a/test/java/util/stream/bootlib/java/util/stream/StreamTestScenario.java Tue Sep 03 11:44:34 2013 -0700 +++ b/test/java/util/stream/bootlib/java/util/stream/StreamTestScenario.java Tue Sep 03 12:16:01 2013 -0700 @@ -39,7 +39,7 @@ @SuppressWarnings({"rawtypes", "unchecked"}) public enum StreamTestScenario implements OpTestCase.BaseStreamTestScenario { - STREAM_FOR_EACH(false) { + STREAM_FOR_EACH_WITH_CLOSE(false) { <T, U, S_IN extends BaseStream<T, S_IN>> void _run(TestData<T, S_IN> data, Consumer<U> b, Function<S_IN, Stream<U>> m) { Stream<U> s = m.apply(data.stream()); @@ -47,6 +47,7 @@ s = s.sequential(); } s.forEach(b); + s.close(); } },
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/test/java/util/stream/test/org/openjdk/tests/java/util/stream/StreamCloseTest.java Tue Sep 03 12:16:01 2013 -0700 @@ -0,0 +1,166 @@ +/* + * Copyright (c) 2013, Oracle and/or its affiliates. All rights reserved. + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. + * + * This code is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License version 2 only, as + * published by the Free Software Foundation. + * + * This code is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * version 2 for more details (a copy is included in the LICENSE file that + * accompanied this code). + * + * You should have received a copy of the GNU General Public License version + * 2 along with this work; if not, write to the Free Software Foundation, + * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. + * + * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA + * or visit www.oracle.com if you need additional information or have any + * questions. + */ +package org.openjdk.tests.java.util.stream; + +import java.util.Arrays; +import java.util.stream.OpTestCase; +import java.util.stream.Stream; + +import org.testng.annotations.Test; + +import static java.util.stream.LambdaTestHelpers.countTo; + +/** + * StreamCloseTest + * + * @author Brian Goetz + */ +@Test(groups = { "serialization-hostile" }) +public class StreamCloseTest extends OpTestCase { + public void testEmptyCloseHandler() { + try (Stream<Integer> ints = countTo(100).stream()) { + ints.forEach(i -> {}); + } + } + + public void testOneCloseHandler() { + final boolean[] holder = new boolean[1]; + Runnable closer = () -> { holder[0] = true; }; + + try (Stream<Integer> ints = countTo(100).stream()) { + ints.onClose(closer); + ints.forEach(i -> {}); + } + assertTrue(holder[0]); + + Arrays.fill(holder, false); + try (Stream<Integer> ints = countTo(100).stream().onClose(closer)) { + ints.forEach(i -> {}); + } + assertTrue(holder[0]); + + Arrays.fill(holder, false); + try (Stream<Integer> ints = countTo(100).stream().filter(e -> true).onClose(closer)) { + ints.forEach(i -> {}); + } + assertTrue(holder[0]); + + Arrays.fill(holder, false); + try (Stream<Integer> ints = countTo(100).stream().filter(e -> true).onClose(closer).filter(e -> true)) { + ints.forEach(i -> {}); + } + assertTrue(holder[0]); + } + + public void testTwoCloseHandlers() { + final boolean[] holder = new boolean[2]; + Runnable close1 = () -> { holder[0] = true; }; + Runnable close2 = () -> { holder[1] = true; }; + + try (Stream<Integer> ints = countTo(100).stream()) { + ints.onClose(close1).onClose(close2); + ints.forEach(i -> {}); + } + assertTrue(holder[0] && holder[1]); + + Arrays.fill(holder, false); + try (Stream<Integer> ints = countTo(100).stream().onClose(close1).onClose(close2)) { + ints.forEach(i -> {}); + } + assertTrue(holder[0] && holder[1]); + + Arrays.fill(holder, false); + try (Stream<Integer> ints = countTo(100).stream().filter(e -> true).onClose(close1).onClose(close2)) { + ints.forEach(i -> {}); + } + assertTrue(holder[0] && holder[1]); + + Arrays.fill(holder, false); + try (Stream<Integer> ints = countTo(100).stream().filter(e -> true).onClose(close1).onClose(close2).filter(e -> true)) { + ints.forEach(i -> {}); + } + assertTrue(holder[0] && holder[1]); + } + + public void testCascadedExceptions() { + final boolean[] holder = new boolean[3]; + boolean caught = false; + Runnable close1 = () -> { holder[0] = true; throw new RuntimeException("1"); }; + Runnable close2 = () -> { holder[1] = true; throw new RuntimeException("2"); }; + Runnable close3 = () -> { holder[2] = true; throw new RuntimeException("3"); }; + + try (Stream<Integer> ints = countTo(100).stream()) { + ints.onClose(close1).onClose(close2).onClose(close3); + ints.forEach(i -> {}); + } + catch (RuntimeException e) { + assertCascaded(e, 3); + assertTrue(holder[0] && holder[1] && holder[2]); + caught = true; + } + assertTrue(caught); + + Arrays.fill(holder, false); + caught = false; + try (Stream<Integer> ints = countTo(100).stream().onClose(close1).onClose(close2).onClose(close3)) { + ints.forEach(i -> {}); + } + catch (RuntimeException e) { + assertCascaded(e, 3); + assertTrue(holder[0] && holder[1] && holder[2]); + caught = true; + } + assertTrue(caught); + + caught = false; + Arrays.fill(holder, false); + try (Stream<Integer> ints = countTo(100).stream().filter(e -> true).onClose(close1).onClose(close2).onClose(close3)) { + ints.forEach(i -> {}); + } + catch (RuntimeException e) { + assertCascaded(e, 3); + assertTrue(holder[0] && holder[1] && holder[2]); + caught = true; + } + assertTrue(caught); + + caught = false; + Arrays.fill(holder, false); + try (Stream<Integer> ints = countTo(100).stream().filter(e -> true).onClose(close1).onClose(close2).filter(e -> true).onClose(close3)) { + ints.forEach(i -> {}); + } + catch (RuntimeException e) { + assertCascaded(e, 3); + assertTrue(holder[0] && holder[1] && holder[2]); + caught = true; + } + assertTrue(caught); + } + + private void assertCascaded(RuntimeException e, int n) { + assertTrue(e.getMessage().equals("1")); + assertTrue(e.getSuppressed().length == n - 1); + for (int i=0; i<n-1; i++) + assertTrue(e.getSuppressed()[i].getMessage().equals(String.valueOf(i + 2))); + } +}