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**

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

@@ -172,7 +172,7 @@ public class StreamingInboundTests {
 		assertNull(out.receive(0));
 
 		// close by list, splitter
-		verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(2)).close();
+		verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(3)).close();
 
 		receivedStream = streamer.receive();
 		splitter.handleMessage(receivedStream);
@@ -187,7 +187,7 @@ public class StreamingInboundTests {
 		assertNull(out.receive(0));
 
 		// close by splitter
-		verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(3)).close();
+		verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(5)).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-2017 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.
@@ -43,6 +43,7 @@ import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.Mockito;
 import org.reactivestreams.Subscriber;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -51,6 +52,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.FluxMessageChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
@@ -75,6 +77,7 @@ import reactor.test.StepVerifier;
 /**
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Ruslan Stelmachenko
  *
  * @since 4.1.2
  */
@@ -365,6 +368,24 @@ public class FileSplitterTests {
 		assertEquals(0, 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")