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 {
+	public void close() {
-		if (this.iterator instanceof Closeable) {
+		if (this.iterator instanceof AutoCloseable) {
-			((Closeable) this.iterator).close();
+			try {
+				((AutoCloseable) this.iterator).close();
+			}
+			catch (Exception e) {
+				// NOSONAR
+			}
+		}
+		if (this.closeable != null) {
+			try {
+				this.closeable.close();
+			}
+			catch (Exception e) {
+				// NOSONAR
+			}
 		}
 	}
 

spring-integration-core/src/test/java/org/springframework/integration/splitter/MethodInvokingSplitterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,15 +18,20 @@ package org.springframework.integration.splitter;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Stream;
 
+import org.aopalliance.intercept.MethodInterceptor;
 import org.junit.Test;
 
+import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.annotation.Splitter;
 import org.springframework.integration.channel.QueueChannel;
@@ -39,6 +44,7 @@ import org.springframework.messaging.support.GenericMessage;
 /**
  * @author Mark Fisher
  * @author Artem Bilan
+ * @author Gary Russell
  */
 public class MethodInvokingSplitterTests {
 
@@ -46,7 +52,7 @@ public class MethodInvokingSplitterTests {
 
 	@Test
 	public void splitStringToStringArray() throws Exception {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = this.getSplitter("stringToStringArray");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -62,7 +68,7 @@ public class MethodInvokingSplitterTests {
 
 	@Test
 	public void splitStringToStringList() throws Exception {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = this.getSplitter("stringToStringList");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -78,7 +84,7 @@ public class MethodInvokingSplitterTests {
 
 	@Test
 	public void splitMessageToStringArray() throws Exception {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = this.getSplitter("messageToStringArray");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -94,7 +100,7 @@ public class MethodInvokingSplitterTests {
 
 	@Test
 	public void splitMessageToStringList() throws Exception {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = this.getSplitter("messageToStringList");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -110,7 +116,7 @@ public class MethodInvokingSplitterTests {
 
 	@Test
 	public void splitMessageToMessageArray() throws Exception {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = this.getSplitter("messageToMessageArray");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -126,7 +132,7 @@ public class MethodInvokingSplitterTests {
 
 	@Test
 	public void splitMessageToMessageList() throws Exception {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = this.getSplitter("messageToMessageList");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -141,7 +147,7 @@ public class MethodInvokingSplitterTests {
 	}
 
 	@Test
-	public void splitMessageToMessageBuilderList() throws Exception {
+	public void splitMessageToMessageBuilderList() {
 		Message<String> message = MessageBuilder.withPayload("foo.bar")
 				.setHeader("myHeader", "myValue")
 				.build();
@@ -165,7 +171,7 @@ public class MethodInvokingSplitterTests {
 
 	@Test
 	public void splitStringToMessageArray() throws Exception {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = this.getSplitter("stringToMessageArray");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -181,7 +187,7 @@ public class MethodInvokingSplitterTests {
 
 	@Test
 	public void splitStringToMessageList() throws Exception {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = this.getSplitter("stringToMessageList");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -196,8 +202,8 @@ public class MethodInvokingSplitterTests {
 	}
 
 	@Test
-	public void splitStringToStringArrayConfiguredByMethodName() throws Exception {
+	public void splitStringToStringArrayConfiguredByMethodName() {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = new MethodInvokingSplitter(testBean, "stringToStringArray");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -212,8 +218,8 @@ public class MethodInvokingSplitterTests {
 	}
 
 	@Test
-	public void splitStringToStringListConfiguredByMethodName() throws Exception {
+	public void splitStringToStringListConfiguredByMethodName() {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = new MethodInvokingSplitter(testBean, "stringToStringList");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -228,8 +234,8 @@ public class MethodInvokingSplitterTests {
 	}
 
 	@Test
-	public void splitMessageToStringArrayConfiguredByMethodName() throws Exception {
+	public void splitMessageToStringArrayConfiguredByMethodName() {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = new MethodInvokingSplitter(testBean, "messageToStringArray");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -244,8 +250,8 @@ public class MethodInvokingSplitterTests {
 	}
 
 	@Test
-	public void splitMessageToStringListConfiguredByMethodName() throws Exception {
+	public void splitMessageToStringListConfiguredByMethodName() {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = new MethodInvokingSplitter(testBean, "messageToStringList");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -260,8 +266,8 @@ public class MethodInvokingSplitterTests {
 	}
 
 	@Test
-	public void splitMessageToMessageArrayConfiguredByMethodName() throws Exception {
+	public void splitMessageToMessageArrayConfiguredByMethodName() {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = new MethodInvokingSplitter(testBean, "messageToMessageArray");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -276,8 +282,8 @@ public class MethodInvokingSplitterTests {
 	}
 
 	@Test
-	public void splitMessageToMessageListConfiguredByMethodName() throws Exception {
+	public void splitMessageToMessageListConfiguredByMethodName() {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = new MethodInvokingSplitter(testBean, "messageToMessageList");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -292,8 +298,8 @@ public class MethodInvokingSplitterTests {
 	}
 
 	@Test
-	public void splitStringToMessageArrayConfiguredByMethodName() throws Exception {
+	public void splitStringToMessageArrayConfiguredByMethodName() {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = new MethodInvokingSplitter(testBean, "stringToMessageArray");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -308,8 +314,8 @@ public class MethodInvokingSplitterTests {
 	}
 
 	@Test
-	public void splitStringToMessageListConfiguredByMethodName() throws Exception {
+	public void splitStringToMessageListConfiguredByMethodName() {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = new MethodInvokingSplitter(testBean, "stringToMessageList");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -324,14 +330,15 @@ public class MethodInvokingSplitterTests {
 	}
 
 	@Test
-	public void splitListPayload() throws Exception {
+	public void splitListPayload() {
 		class ListSplitter {
+
 			@SuppressWarnings("unused")
 			public List<String> split(List<String> list) {
 				return list;
 			}
 		}
-		GenericMessage<List<?>> message = new GenericMessage<List<?>>(Arrays.asList("foo", "bar"));
+		GenericMessage<List<?>> message = new GenericMessage<>(Arrays.asList("foo", "bar"));
 		MethodInvokingSplitter splitter = new MethodInvokingSplitter(new ListSplitter(), "split");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -345,9 +352,46 @@ public class MethodInvokingSplitterTests {
 		assertEquals("bar", reply2.getPayload());
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	public void splitStreamPayload() {
+		class StreamSplitter {
+
+			@SuppressWarnings("unused")
+			public Stream<String> split(Stream<String> stream) {
+				return stream;
+			}
+
+		}
+		Stream<String> stream = Stream.of("foo", "bar");
+		ProxyFactory pf = new ProxyFactory(stream);
+		AtomicBoolean closed = new AtomicBoolean();
+		MethodInterceptor interceptor = i -> {
+			if (i.getMethod().getName().equals("close")) {
+				closed.set(true);
+			}
+			return i.proceed();
+		};
+		pf.addAdvice(interceptor);
+		stream = (Stream<String>) pf.getProxy();
+		GenericMessage<Stream<?>> message = new GenericMessage<>(stream);
+		MethodInvokingSplitter splitter = new MethodInvokingSplitter(new StreamSplitter(), "split");
+		QueueChannel replyChannel = new QueueChannel();
+		splitter.setOutputChannel(replyChannel);
+		splitter.handleMessage(message);
+		List<Message<?>> replies = replyChannel.clear();
+		Message<?> reply1 = replies.get(0);
+		assertNotNull(reply1);
+		assertEquals("foo", reply1.getPayload());
+		Message<?> reply2 = replies.get(1);
+		assertNotNull(reply2);
+		assertEquals("bar", reply2.getPayload());
+		assertTrue("Expected stream.close()", closed.get());
+	}
+
 	@Test
 	public void headerForObjectReturnValues() throws Exception {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = this.getSplitter("stringToStringArray");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -367,7 +411,7 @@ public class MethodInvokingSplitterTests {
 
 	@Test
 	public void headerForMessageReturnValues() throws Exception {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		MethodInvokingSplitter splitter = this.getSplitter("messageToMessageList");
 		QueueChannel replyChannel = new QueueChannel();
 		splitter.setOutputChannel(replyChannel);
@@ -427,7 +471,7 @@ public class MethodInvokingSplitterTests {
 
 	@Test
 	public void singleAnnotation() {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		SingleAnnotationTestBean annotatedBean = new SingleAnnotationTestBean();
 		MethodInvokingSplitter splitter = new MethodInvokingSplitter(annotatedBean);
 		QueueChannel replyChannel = new QueueChannel();
@@ -449,7 +493,7 @@ public class MethodInvokingSplitterTests {
 
 	@Test
 	public void singlePublicMethod() {
-		GenericMessage<String> message = new GenericMessage<String>("foo.bar");
+		GenericMessage<String> message = new GenericMessage<>("foo.bar");
 		SinglePublicMethodTestBean testBean = new SinglePublicMethodTestBean();
 		MethodInvokingSplitter splitter = new MethodInvokingSplitter(testBean);
 		QueueChannel replyChannel = new QueueChannel();
@@ -504,7 +548,7 @@ public class MethodInvokingSplitterTests {
 
 		public List<Message<String>> messageToMessageList(Message<?> input) {
 			String[] strings = input.getPayload().toString().split("\\.");
-			List<Message<String>> messages = new ArrayList<Message<String>>();
+			List<Message<String>> messages = new ArrayList<>();
 			for (String s : strings) {
 				messages.add(new GenericMessage<String>(s));
 			}
@@ -533,9 +577,9 @@ public class MethodInvokingSplitterTests {
 
 		public List<Message<String>> stringToMessageList(String input) {
 			String[] strings = input.split("\\.");
-			List<Message<String>> messages = new ArrayList<Message<String>>();
+			List<Message<String>> messages = new ArrayList<>();
 			for (String s : strings) {
-				messages.add(new GenericMessage<String>(s));
+				messages.add(new GenericMessage<>(s));
 			}
 			return messages;
 		}
@@ -546,7 +590,7 @@ public class MethodInvokingSplitterTests {
 
 		public List<String> splitPayloadAndHeader(String payload, @Header("testHeader") String header) {
 			String regex = "\\.";
-			List<String> results = new ArrayList<String>();
+			List<String> results = new ArrayList<>();
 			Collections.addAll(results, payload.split(regex));
 			Collections.addAll(results, header.split(regex));
 			return results;
@@ -614,4 +658,5 @@ public class MethodInvokingSplitterTests {
 		}
 
 	}
+
 }

spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -329,7 +329,7 @@ public class XPathMessageSplitter extends AbstractMessageSplitter {
 
 		TransformFunctionIterator(Iterator<Node> delegate,
 				Function<? super Node, ? extends String> function) {
-			super(delegate, function);
+			super(null, delegate, function);
 			this.delegate = delegate;
 		}