changeset 19800:6e1fef53ea55

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
author henryjen
date Tue, 03 Sep 2013 12:16:01 -0700
parents efa9ff09b024
children 97f3caac6c77
files jdk/src/share/classes/java/nio/file/Files.java jdk/src/share/classes/java/util/stream/AbstractPipeline.java jdk/src/share/classes/java/util/stream/BaseStream.java jdk/src/share/classes/java/util/stream/CloseableStream.java jdk/src/share/classes/java/util/stream/DelegatingStream.java jdk/src/share/classes/java/util/stream/DoublePipeline.java jdk/src/share/classes/java/util/stream/DoubleStream.java jdk/src/share/classes/java/util/stream/IntPipeline.java jdk/src/share/classes/java/util/stream/IntStream.java jdk/src/share/classes/java/util/stream/LongPipeline.java jdk/src/share/classes/java/util/stream/LongStream.java jdk/src/share/classes/java/util/stream/ReferencePipeline.java jdk/src/share/classes/java/util/stream/Stream.java jdk/src/share/classes/java/util/stream/Streams.java jdk/test/java/nio/file/Files/StreamTest.java jdk/test/java/util/stream/bootlib/java/util/stream/DoubleStreamTestScenario.java jdk/test/java/util/stream/bootlib/java/util/stream/IntStreamTestScenario.java jdk/test/java/util/stream/bootlib/java/util/stream/LongStreamTestScenario.java jdk/test/java/util/stream/bootlib/java/util/stream/StreamTestScenario.java jdk/test/java/util/stream/test/org/openjdk/tests/java/util/stream/StreamCloseTest.java
diffstat 20 files changed, 579 insertions(+), 549 deletions(-) [+]
line wrap: on
line diff
--- a/jdk/src/share/classes/java/nio/file/Files.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/src/share/classes/java/util/stream/AbstractPipeline.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/src/share/classes/java/util/stream/BaseStream.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/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/jdk/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/jdk/src/share/classes/java/util/stream/DoublePipeline.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/src/share/classes/java/util/stream/DoubleStream.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/src/share/classes/java/util/stream/IntPipeline.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/src/share/classes/java/util/stream/IntStream.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/src/share/classes/java/util/stream/LongPipeline.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/src/share/classes/java/util/stream/LongStream.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/src/share/classes/java/util/stream/ReferencePipeline.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/src/share/classes/java/util/stream/Stream.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/src/share/classes/java/util/stream/Streams.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/test/java/nio/file/Files/StreamTest.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/test/java/util/stream/bootlib/java/util/stream/DoubleStreamTestScenario.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/test/java/util/stream/bootlib/java/util/stream/IntStreamTestScenario.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/test/java/util/stream/bootlib/java/util/stream/LongStreamTestScenario.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/test/java/util/stream/bootlib/java/util/stream/StreamTestScenario.java	Tue Sep 03 11:44:34 2013 -0700
+++ b/jdk/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/jdk/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)));
+    }
+}