INT-4506: Splitter: close stream and Closeables

JIRA: https://jira.spring.io/browse/INT-4506

Capture the actual splitter result in the `FunctionIterator and close it if necessary.

**cherry-pick to 5.0.x**

Add Closeable to proxy in test; remove AutoCloseable check.

Closeable extends AutoCloseable.

Fix javadocs; test for AutoCloseable instead of Closeable

Polishing - PR comments; tighten up API.

A `CloseableIterator` is now an `AutoCloseable` with an overridden
`close()`` (no exception) to avoid lint problem.

* Polishing omissions, code style and `@author`

**Cherry-pick to 5.0.x**

spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java

@@ -16,7 +16,6 @@
 
 package org.springframework.integration.splitter;
 
-import java.io.Closeable;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -44,6 +43,7 @@ import reactor.core.publisher.Flux;
  * @author Dave Syer
  * @author Artem Bilan
  * @author Ruslan Stelmachenko
+ * @author Gary Russell
  */
 public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler {
 
@@ -155,7 +155,8 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 			return flux.map(messageBuilderFunction);
 		}
 		else {
-			return new FunctionIterator<>(iterator, messageBuilderFunction);
+			return new FunctionIterator<>(result instanceof AutoCloseable && !result.equals(iterator)
+					? (AutoCloseable) result : null, iterator, messageBuilderFunction);
 		}
 	}
 
@@ -235,9 +236,9 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 				}
 			}
 			finally {
-				if (iterator instanceof Closeable) {
+				if (iterator instanceof AutoCloseable) {
 					try {
-						((Closeable) iterator).close();
+						((AutoCloseable) iterator).close();
 					}
 					catch (Exception e) {
 						// ignored

spring-integration-core/src/main/java/org/springframework/integration/util/CloseableIterator.java

@@ -16,7 +16,6 @@
 
 package org.springframework.integration.util;
 
-import java.io.Closeable;
 import java.util.Iterator;
 
 /**
@@ -24,8 +23,13 @@ import java.util.Iterator;
  * This allows implementations to clean up any resources they need to keep open to iterate over elements.
  *
  * @author Ruslan Stelmachenko
+ * @author Gary Russell
  *
  * @since 4.3.15
  */
-public interface CloseableIterator<E> extends Iterator<E>, Closeable {
+public interface CloseableIterator<E> extends Iterator<E>, AutoCloseable {
+
+	@Override
+	void close(); // override throws Exception
+
 }

spring-integration-core/src/main/java/org/springframework/integration/util/FunctionIterator.java

@@ -16,39 +16,74 @@
 
 package org.springframework.integration.util;
 
-import java.io.Closeable;
-import java.io.IOException;
 import java.util.Iterator;
 import java.util.function.Function;
 
+import org.springframework.lang.Nullable;
+
 /**
  * An {@link Iterator} implementation to convert each item from the target
  * {@link #iterator} to a new object applying the {@link #function} on {@link #next()}.
  *
  * @author Artem Bilan
  * @author Ruslan Stelmachenko
+ * @author Gary Russell
  * @since 4.1
  */
 public class FunctionIterator<T, V> implements CloseableIterator<V> {
 
+	private final AutoCloseable closeable;
+
 	private final Iterator<T> iterator;
 
 	private final Function<? super T, ? extends V> function;
 
+	/**
+	 * Construct an instance with the provided iterable and function.
+	 * @param iterable the iterable.
+	 * @param function the function.
+	 */
 	public FunctionIterator(Iterable<T> iterable, Function<? super T, ? extends V> function) {
-		this(iterable.iterator(), function);
+		this(null, iterable.iterator(), function);
 	}
 
+	/**
+	 * Construct an instance with the provided root object, iterable and function.
+	 * @param closeable an {@link AutoCloseable} to close when iteration is complete.
+	 * @param iterable the iterable.
+	 * @param function the function.
+	 * @since 5.0.7
+	 */
+	public FunctionIterator(@Nullable AutoCloseable closeable, Iterable<T> iterable,
+			Function<? super T, ? extends V> function) {
+
+		this(closeable, iterable.iterator(), function);
+	}
+
+	/**
+	 * Construct an instance with the provided iterator and function.
+	 * @param newIterator the iterator.
+	 * @param function the function.
+	 */
 	public FunctionIterator(Iterator<T> newIterator, Function<? super T, ? extends V> function) {
+		this(null, newIterator, function);
+	}
+
+	/**
+	 * Construct an instance with the provided root object, iterator and function.
+	 * @param closeable an {@link AutoCloseable} to close when iteration is complete.
+	 * @param newIterator the iterator.
+	 * @param function the function.
+	 * @since 5.0.7
+	 */
+	public FunctionIterator(@Nullable AutoCloseable closeable, Iterator<T> newIterator,
+			Function<? super T, ? extends V> function) {
+
+		this.closeable = closeable;
 		this.iterator = newIterator;
 		this.function = function;
 	}
 
-	@Override
-	public void remove() {
-		throw new UnsupportedOperationException("Cannot remove from a collect iterator");
-	}
-
 	@Override
 	public boolean hasNext() {
 		return this.iterator.hasNext();
@@ -60,9 +95,22 @@ public class FunctionIterator<T, V> implements CloseableIterator<V> {
 	}
 
 	@Override
-	public void close() throws IOException {
-		if (this.iterator instanceof Closeable) {
-			((Closeable) this.iterator).close();
+	public void close() {
+		if (this.iterator instanceof AutoCloseable) {
+			try {
+				((AutoCloseable) this.iterator).close();
+			}
+			catch (Exception e) {
+				// NOSONAR
+			}
+		}
+		if (this.closeable != null) {
+			try {
+				this.closeable.close();
+			}
+			catch (Exception e) {
+				// NOSONAR
+			}
 		}
 	}