Additional changes to IOUtils close and exception handling

src/main/java/org/apache/commons/io/IOUtils.java

@@ -45,11 +45,11 @@
 import java.nio.charset.Charset;
 import java.nio.file.Files;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Consumer;
+import java.util.stream.Stream;
 
 import org.apache.commons.io.function.IOConsumer;
 import org.apache.commons.io.input.QueueInputStream;
@@ -394,6 +394,25 @@ public static void close(final Closeable... closeables) throws IOException {
         IOConsumer.forEach(closeables, IOUtils::close);
     }
 
+    /**
+     * Closes the entries in the given {@link Stream<T>} as null-safe operations,
+     * and closes the underlying {@code Stream}.
+     *
+     * @param <T> The element type.
+     * @param closeables The resource(s) to close, may be null.
+     * @throws IOExceptionList if an I/O error occurs.
+     * @since 2.12.0
+     */
+    public static <T extends Closeable> void close(final Stream<T> closeables) throws IOExceptionList {
+        if (closeables != null) {
+            try {
+                IOConsumer.forEachIndexed(closeables, IOUtils::close);
+            } finally {
+                closeables.close();
+            }
+        }
+    }
+
     /**
      * Closes the given {@link Closeable} as a null-safe operation.
      *
@@ -415,13 +434,36 @@ public static void close(final Closeable closeable, final IOConsumer<IOException
     }
 
     /**
-     * Closes the given {@link Closeable} as a null-safe operation.
+     * Closes the entries in the given {@link Stream<T>} as null-safe operations,
+     * and closes the underlying {@code Stream}.
      *
+     * @param <T> The element type.
+     * @param consumer Consume the IOException thrown by {@link Closeable#close()}.
      * @param closeables The resource(s) to close, may be null.
+     * @throws IOException if an I/O error occurs.
+     */
+    public static <T extends Closeable> void close(final IOConsumer<IOException> consumer, final Stream<T> closeables) throws IOException {
+        if (closeables != null) {
+            try {
+                close(closeables);
+            } catch (final IOException e) {
+                if (consumer != null) {
+                    consumer.accept(e);
+                }
+            } finally {
+                closeables.close();
+            }
+        }
+    }
+
+    /**
+     * Closes the given {@link Closeable} as a null-safe operation.
+     *
      * @param consumer Consume the IOException thrown by {@link Closeable#close()}.
+     * @param closeables The resource(s) to close, may be null.
      * @throws IOException if an I/O error occurs.
      */
-    public static void close(final Closeable[] closeables, final IOConsumer<IOException> consumer) throws IOException {
+    public static void close(final IOConsumer<IOException> consumer, final Closeable... closeables) throws IOException {
         if (closeables != null) {
             try {
                 close(closeables);
@@ -538,7 +580,7 @@ public static void closeQuietly(final Closeable closeable) {
      */
     public static void closeQuietly(final Closeable... closeables) {
         if (closeables != null) {
-            Arrays.stream(closeables).forEach(IOUtils::closeQuietly);
+            IOConsumer.forEachQuietly(closeables, IOUtils::closeQuietly);
         }
     }
 
@@ -561,6 +603,64 @@ public static void closeQuietly(final Closeable closeable, final Consumer<IOExce
         }
     }
 
+    /**
+     * Closes the given {@link Stream<T>} as a null-safe operation while consuming IOException by the given {@code consumer},
+     * and closes the underlying {@code Stream}.
+     *
+     * @param <T> The element type.
+     * @param closeables The resource(s) to close, may be null.
+     * @since 2.12.0
+     */
+    public static <T extends Closeable> void closeQuietly(final Stream<T> closeables) {
+        closeQuietly(null, closeables);
+    }
+
+    /**
+     * Closes the given {@link Stream<T>} as a null-safe operation while consuming IOException by the given {@code consumer},
+     * and closes the underlying {@code Stream}.
+     *
+     * @param <T> The element type.
+     * @param consumer Consume the IOException thrown by {@link Closeable#close()}.
+     * @param closeables The resource(s) to close, may be null.
+     * @since 2.12.0
+     */
+    public static <T extends Closeable> void closeQuietly(final Consumer<IOException> consumer, final Stream<T> closeables) {
+        if (closeables != null) {
+            try {
+                close(closeables);
+            } catch (final IOException e) {
+                if (consumer != null) {
+                    consumer.accept(e);
+                }
+            } finally {
+                try {
+                    closeables.close();
+                } catch (Exception e) {
+                    // Do nothing.
+                }
+            }
+        }
+    }
+
+    /**
+     * Closes the given {@link Closeable}s as a null-safe operation while consuming IOException by the given {@code consumer}.
+     *
+     * @param consumer Consume the IOException thrown by {@link Closeable#close()}.
+     * @param closeables The resource(s) to close, may be null.
+     * @since 2.12.0
+     */
+    public static void closeQuietly(final Consumer<IOException> consumer, final Closeable... closeables) {
+        if (closeables != null) {
+            try {
+                close(closeables);
+            } catch (final IOException e) {
+                if (consumer != null) {
+                    consumer.accept(e);
+                }
+            }
+        }
+    }
+
     /**
      * Closes an {@code InputStream} unconditionally.
      * <p>

src/main/java/org/apache/commons/io/function/IOConsumer.java

@@ -18,7 +18,9 @@
 package org.apache.commons.io.function;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 
@@ -39,6 +41,29 @@ public interface IOConsumer<T> {
      */
     IOConsumer<?> NOOP_IO_CONSUMER = t -> {/* noop */};
 
+    /**
+     * Wraps an {@code IOConsumer} inside of a {@link Consumer}
+     * that throws {@link UncheckedIOException} for any {@link IOException}s
+     * that are thrown by the underlying {@code IOConsumer}.
+     *
+     * @param <T> The element type.
+     * @param consumer The {@code IOConsumer} to wrap.
+     * @return a {@code Consumer} that wraps the given {@code IOConsumer}.
+     * @since 2.12.0
+     */
+    static <T> Consumer<T> wrap(IOConsumer<T> consumer) {
+        return new Consumer<T>() {
+            @Override
+            public void accept(T t) {
+                try {
+                    consumer.accept(t);
+                } catch (IOException e) {
+                    throw new UncheckedIOException(e);
+                }
+            }
+        };
+    }
+
     /**
      * Performs an action for each element of this stream.
      *
@@ -52,6 +77,46 @@ static <T> void forEach(final T[] array, final IOConsumer<T> action) throws IOEx
         IOStreams.forEach(IOStreams.of(array), action);
     }
 
+    /**
+     * Performs an action for each element of this array, returning
+     * a {@link Optional} that either contains an {@link IOException}
+     * if one occurred, or {@link Optional#empty()}.
+     *
+     * @param <T> The element type.
+     * @param array The input to stream.
+     * @param action The action to apply to each input element.
+     * @return a {@code Optional} that may wrap a {@code IOException}.
+     * @since 2.12.0
+     */
+    static <T> Optional<IOException> forEachQuietly(final T[] array, final IOConsumer<T> action) {
+        try {
+            IOStreams.forEach(IOStreams.of(array), action);
+            return Optional.empty();
+        } catch (IOException e) {
+            return Optional.of(e);
+        }
+    }
+
+    /**
+     * Performs an action for each element of this stream, returning
+     * a {@link Optional} that either contains an {@link IOExceptionList}
+     * if one occurred, or {@link Optional#empty()}.
+     *
+     * @param <T> The element type.
+     * @param stream The input to stream.
+     * @param action The action to apply to each input element.
+     * @return a {@code Optional} that may wrap a {@code IOExceptionList}.
+     * @since 2.12.0
+     */
+    static <T> Optional<IOExceptionList> forEachIndexedQuietly(final Stream<T> stream, final IOConsumer<T> action) {
+        try {
+            IOStreams.forEachIndexed(stream, action, IOIndexedException::new);
+            return Optional.empty();
+        } catch (IOExceptionList e) {
+            return Optional.of(e);
+        }
+    }
+
     /**
      * Performs an action for each element of this stream.
      *
@@ -85,13 +150,25 @@ static <T> IOConsumer<T> noop() {
      */
     void accept(T t) throws IOException;
 
+    /**
+     * Returns this {@code IOConsumer} wrapped inside of a {@link Consumer}
+     * that throws {@link UncheckedIOException} for any {@link IOException}s
+     * that are thrown by this {@code IOConsumer}.
+     *
+     * @return a {@code Consumer} that wraps this {@code IOConsumer}.
+     * @since 2.12.0
+     */
+    default Consumer<T> asConsumer() {
+        return wrap(this);
+    }
+
     /**
      * Returns a composed {@code IOConsumer} that performs, in sequence, this operation followed by the {@code after}
      * operation. If performing either operation throws an exception, it is relayed to the caller of the composed operation.
      * If performing this operation throws an exception, the {@code after} operation will not be performed.
      *
      * @param after the operation to perform after this operation
-     * @return a composed {@code Consumer} that performs in sequence this operation followed by the {@code after} operation
+     * @return a composed {@code IOConsumer} that performs in sequence this operation followed by the {@code after} operation
      * @throws NullPointerException if {@code after} is null
      */
     default IOConsumer<T> andThen(final IOConsumer<? super T> after) {

src/main/java/org/apache/commons/io/function/IOStreams.java

@@ -52,20 +52,27 @@ static <T> void forEach(final Stream<T> stream, final IOConsumer<T> action) thro
 
     static <T> void forEachIndexed(final Stream<T> stream, final IOConsumer<T> action, final BiFunction<Integer, IOException, IOException> exSupplier)
         throws IOExceptionList {
-        final AtomicReference<List<IOException>> causeList = new AtomicReference<>();
+        final AtomicReference<List<Throwable>> causeList = new AtomicReference<>();
         final AtomicInteger index = new AtomicInteger();
-        stream.forEach(e -> {
-            try {
-                action.accept(e);
-            } catch (final IOException ioex) {
-                if (causeList.get() == null) {
-                    causeList.set(new ArrayList<>());
+        try {
+            stream.forEach(e -> {
+                try {
+                    action.accept(e);
+                } catch (final IOException ioex) {
+                    if (causeList.get() == null) {
+                        causeList.set(new ArrayList<>());
+                    }
+                    causeList.get().add(exSupplier.apply(index.get(), ioex));
                 }
-                causeList.get().add(exSupplier.apply(index.get(), ioex));
+                index.incrementAndGet();
+            });
+        }
+        catch (Throwable t) {
+            if (causeList.get() == null) {
+                causeList.set(new ArrayList<>());
             }
-            index.incrementAndGet();
-        });
+            causeList.get().add(t);
+        }
         IOExceptionList.checkEmpty(causeList.get(), "forEach");
     }
-
 }

src/main/java/org/apache/commons/io/input/ObservableInputStream.java

@@ -186,7 +186,7 @@ public List<Observer> getObservers() {
     }
 
     /**
-     * Notifies the observers by invoking {@link Observer#finished()}.
+     * Notifies the observers by invoking {@link Observer#closed()}.
      *
      * @throws IOException Some observer has thrown an exception, which is being passed down.
      */

src/main/java/org/apache/commons/io/output/FilterCollectionWriter.java

@@ -28,6 +28,7 @@
 
 import org.apache.commons.io.IOExceptionList;
 import org.apache.commons.io.IOIndexedException;
+import org.apache.commons.io.IOUtils;
 import org.apache.commons.io.function.IOConsumer;
 
 /**
@@ -102,7 +103,7 @@ public Writer append(final CharSequence csq, final int start, final int end) thr
 
     @Override
     public void close() throws IOException {
-        IOConsumer.forEachIndexed(writers(), Writer::close);
+        IOUtils.close(writers());
     }
 
     /**

src/test/java/org/apache/commons/io/IOUtilsTest.java

@@ -20,6 +20,7 @@
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNotSame;
 import static org.junit.jupiter.api.Assertions.assertSame;
@@ -57,8 +58,9 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
-
 import org.apache.commons.io.function.IOConsumer;
 import org.apache.commons.io.input.CircularInputStream;
 import org.apache.commons.io.input.NullInputStream;
@@ -67,7 +69,9 @@
 import org.apache.commons.io.output.NullOutputStream;
 import org.apache.commons.io.output.StringBuilderWriter;
 import org.apache.commons.io.test.TestUtils;
+import org.apache.commons.io.test.ThrowOnCloseInputStream;
 import org.apache.commons.io.test.ThrowOnCloseReader;
+import org.apache.commons.io.test.ThrowOnCloseWriter;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
@@ -380,6 +384,22 @@ public void testCloseMulti() {
             () -> IOUtils.close(nullCloseable, new ThrowOnCloseReader(new StringReader("s"))));
     }
 
+    @Test
+    public void testCloseMultiConsumer() {
+        final Collection<IOException> exceptionCollection = new HashSet<>();
+        final IOConsumer<IOException> checkConsumer = i -> {
+            exceptionCollection.add(i);
+        };
+
+        final Closeable[] closeables = {null, new ThrowOnCloseInputStream(), new ThrowOnCloseReader(), new ThrowOnCloseWriter()};
+        assertDoesNotThrow(() -> IOUtils.close(checkConsumer, closeables));
+        assertEquals(exceptionCollection.size(), 1);
+
+        final IOException exception = exceptionCollection.iterator().next();
+        assertInstanceOf(IOExceptionList.class, exception);
+        assertEquals(((IOExceptionList)exception).getCauseList().size(), 3);
+    }
+
     @Test
     public void testCloseQuietly_AllCloseableIOException() {
         final Closeable closeable = () -> {