diff --git a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java index 3a929c5b0b..028b76283f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.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. @@ -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; @@ -42,6 +43,7 @@ import reactor.core.publisher.Flux; * @author Mark Fisher * @author Dave Syer * @author Artem Bilan + * @author Ruslan Stelmachenko */ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler { @@ -227,9 +229,20 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess protected void produceOutput(Object result, Message requestMessage) { if (result instanceof Iterator) { Iterator iterator = (Iterator) result; - while (iterator.hasNext()) { - super.produceOutput(iterator.next(), requestMessage); - + try { + while (iterator.hasNext()) { + super.produceOutput(iterator.next(), requestMessage); + } + } + finally { + if (iterator instanceof Closeable) { + try { + ((Closeable) iterator).close(); + } + catch (Exception e) { + // ignored + } + } } } else { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/CloseableIterator.java b/spring-integration-core/src/main/java/org/springframework/integration/util/CloseableIterator.java new file mode 100644 index 0000000000..00b4a63ced --- /dev/null +++ b/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 extends Iterator, Closeable { +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/FunctionIterator.java b/spring-integration-core/src/main/java/org/springframework/integration/util/FunctionIterator.java index 021398a5e3..d4ed6f0343 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/FunctionIterator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/FunctionIterator.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 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; import java.util.function.Function; @@ -24,9 +26,10 @@ import java.util.function.Function; * {@link #iterator} to a new object applying the {@link #function} on {@link #next()}. * * @author Artem Bilan + * @author Ruslan Stelmachenko * @since 4.1 */ -public class FunctionIterator implements Iterator { +public class FunctionIterator implements CloseableIterator { private final Iterator iterator; @@ -56,5 +59,11 @@ public class FunctionIterator implements Iterator { return this.function.apply(this.iterator.next()); } -} + @Override + public void close() throws IOException { + if (this.iterator instanceof Closeable) { + ((Closeable) this.iterator).close(); + } + } +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java index 56c9c5be6e..62529c1162 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.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. @@ -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; @@ -68,6 +69,7 @@ import org.springframework.util.StringUtils; * * @author Artem Bilan * @author Gary Russell + * @author Ruslan Stelmachenko * * @since 4.1.2 */ @@ -243,7 +245,7 @@ public class FileSplitter extends AbstractMessageSplitter { firstLineAsHeader = null; } - Iterator iterator = new Iterator() { + Iterator iterator = new CloseableIterator() { boolean markers = FileSplitter.this.markers; @@ -263,7 +265,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; @@ -280,8 +282,8 @@ public class FileSplitter extends AbstractMessageSplitter { } catch (IOException e) { try { - bufferedReader.close(); this.done = true; + bufferedReader.close(); } catch (IOException e1) { // ignored @@ -344,6 +346,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) { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java index 8df56b2227..e37c62871d 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java +++ b/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 { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java index 4aed0f3fef..2623d9ae07 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java +++ b/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(fileReader)); + } + catch (RuntimeException e) { + // ignore + } + Mockito.verify(fileReader).close(); + } + @Configuration @EnableIntegration @ImportResource("classpath:org/springframework/integration/file/splitter/FileSplitterTests-context.xml")