INT-4430: FileSplitter close reader on exception

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

When Iterator-based `FileSplitter` splits the file, and an exception
throws in downstream flow (in the same thread), the exception propagates
to the caller leaving underlying file reader opened.

This commit changes `AbstractMessageSplitter` the way, that,
when any exception happens, if Iterator implements `java.io.Closeable`,
its `close()` method will be called before propagating exception.

Also `FileSplitter`'s underlying iterator implements `Closeable` now.

* Make `CloseableIterator` to follow `Closeable` contract.

Now `CloseableIterator.close()` declares `IOException` and can be used
as base interface for `FunctionIterator`.

* Adjust tests.

Adjust tests to reflect the fact that we call `close()` on the reader
one more time in the end of iterator.

**Cherry-pick to 5.0.x and 4.3.x**

(cherry picked from commit 7f25cba)

# Conflicts:
#	spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
#	spring-integration-core/src/main/java/org/springframework/integration/util/FunctionIterator.java
#	spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java
#	spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java
#	spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.splitter;
 
+import java.io.Closeable;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -36,6 +37,7 @@ import org.springframework.messaging.Message;
  * @author Mark Fisher
  * @author Dave Syer
  * @author Artem Bilan
+ * @author Ruslan Stelmachenko
  */
 public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler {
 
@@ -154,10 +156,26 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 
 	@Override
 	protected void produceOutput(Object result, Message<?> requestMessage) {
-		Iterator<?> iterator = (Iterator<?>) result;
-		while (iterator.hasNext()) {
-			super.produceOutput(iterator.next(), requestMessage);
-
+		if (result instanceof Iterator<?>) {
+			Iterator<?> iterator = (Iterator<?>) result;
+			try {
+				while (iterator.hasNext()) {
+					super.produceOutput(iterator.next(), requestMessage);
+				}
+			}
+			finally {
+				if (iterator instanceof Closeable) {
+					try {
+						((Closeable) iterator).close();
+					}
+					catch (Exception e) {
+						// ignored
+					}
+				}
+			}
+		}
+		else {
+			super.produceOutput(result, requestMessage);
 		}
 	}
 

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

@@ -0,0 +1,31 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.util;
+
+import java.io.Closeable;
+import java.util.Iterator;
+
+/**
+ * A {@link CloseableIterator} is intended to be used when it may hold resources (such as file or socket handles).
+ * This allows implementations to clean up any resources they need to keep open to iterate over elements.
+ *
+ * @author Ruslan Stelmachenko
+ *
+ * @since 4.3.15
+ */
+public interface CloseableIterator<E> extends Iterator<E>, Closeable {
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-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.
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.util;
 
+import java.io.Closeable;
+import java.io.IOException;
 import java.util.Iterator;
 
 /**
@@ -23,9 +25,10 @@ import java.util.Iterator;
  * {@link #iterator} to a new object applying the {@link #function} on {@link #next()}.
  *
  * @author Artem Bilan
+ * @author Ruslan Stelmachenko
  * @since 4.1
  */
-public final class FunctionIterator<T, V> implements Iterator<V> {
+public final class FunctionIterator<T, V> implements CloseableIterator<V> {
 
 	private final Iterator<T> iterator;
 
@@ -55,5 +58,11 @@ public final class FunctionIterator<T, V> implements Iterator<V> {
 		return this.function.apply(this.iterator.next());
 	}
 
-}
+	@Override
+	public void close() throws IOException {
+		if (this.iterator instanceof Closeable) {
+			((Closeable) this.iterator).close();
+		}
+	}
 
+}

spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-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.
@@ -41,6 +41,7 @@ import org.springframework.integration.splitter.AbstractMessageSplitter;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.json.JsonObjectMapper;
 import org.springframework.integration.support.json.JsonObjectMapperProvider;
+import org.springframework.integration.util.CloseableIterator;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.util.Assert;
@@ -61,6 +62,8 @@ import org.springframework.util.StringUtils;
  *
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Ruslan Stelmachenko
+ *
  * @since 4.1.2
  */
 public class FileSplitter extends AbstractMessageSplitter {
@@ -208,7 +211,7 @@ public class FileSplitter extends AbstractMessageSplitter {
 
 		};
 
-		Iterator<Object> iterator = new Iterator<Object>() {
+		Iterator<Object> iterator = new CloseableIterator<Object>() {
 
 			boolean markers = FileSplitter.this.markers;
 
@@ -228,7 +231,7 @@ public class FileSplitter extends AbstractMessageSplitter {
 			public boolean hasNext() {
 				this.hasNextCalled = true;
 				try {
-					if (this.line == null && !this.done) {
+					if (!this.done && this.line == null) {
 						this.line = bufferedReader.readLine();
 					}
 					boolean ready = !this.done && this.line != null;
@@ -245,8 +248,8 @@ public class FileSplitter extends AbstractMessageSplitter {
 				}
 				catch (IOException e) {
 					try {
-						bufferedReader.close();
 						this.done = true;
+						bufferedReader.close();
 					}
 					catch (IOException e1) {
 						// ignored
@@ -300,6 +303,17 @@ public class FileSplitter extends AbstractMessageSplitter {
 						.setHeader(FileHeaders.MARKER, fileMarker.mark.name());
 			}
 
+			@Override
+			public void close() {
+				try {
+					this.done = true;
+					bufferedReader.close();
+				}
+				catch (IOException e) {
+					// ignored
+				}
+			}
+
 		};
 
 		if (this.iterator) {

spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java

@@ -22,6 +22,7 @@ import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willReturn;
 import static org.mockito.BDDMockito.willThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import java.io.ByteArrayInputStream;
@@ -116,7 +117,9 @@ public class StreamingInboundTests {
 		assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE));
 		assertNull(out.receive(0));
 
-		verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource()).close();
+
+		// close by list, splitter
+		verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(2)).close();
 
 		receivedStream = streamer.receive();
 		splitter.handleMessage(receivedStream);
@@ -130,7 +133,9 @@ public class StreamingInboundTests {
 		assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE));
 		assertNull(out.receive(0));
 
-		verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource()).close();
+
+		// close by splitter
+		verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(2)).close();
 	}
 
 	public static class Streamer extends AbstractRemoteFileStreamingMessageSource<String> {

spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-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.
@@ -39,6 +39,7 @@ import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.Mockito;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -46,6 +47,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.ImportResource;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.annotation.Splitter;
+import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.file.FileHeaders;
@@ -66,6 +68,8 @@ import org.springframework.util.FileCopyUtils;
 /**
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Ruslan Stelmachenko
+ *
  * @since 4.1.2
  */
 @ContextConfiguration(loader = AnnotationConfigContextLoader.class)
@@ -264,6 +268,24 @@ public class FileSplitterTests {
 		assertEquals(2, fileMarker.getLineCount());
 	}
 
+	@Test
+	public void testFileReaderClosedOnException() throws Exception {
+		DirectChannel outputChannel = new DirectChannel();
+		outputChannel.subscribe(message -> {
+			throw new RuntimeException();
+		});
+		FileSplitter splitter = new FileSplitter(true, true);
+		splitter.setOutputChannel(outputChannel);
+		FileReader fileReader = Mockito.spy(new FileReader(file));
+		try {
+			splitter.handleMessage(new GenericMessage<Reader>(fileReader));
+		}
+		catch (RuntimeException e) {
+			// ignore
+		}
+		Mockito.verify(fileReader).close();
+	}
+
 	@Configuration
 	@EnableIntegration
 	@ImportResource("classpath:org/springframework/integration/file/splitter/FileSplitterTests-context.xml")