Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4430: FileSplitter close reader on exception

Browse Source 
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**
This commit is contained in:
Ruslan Stelmachenko
2018-03-16 16:26:41 +02:00
committed by Artem Bilan
parent f77fd78fb3
commit 7f25cba262
6 changed files with 101 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 {

31
spring-integration-core/src/main/java/org/springframework/integration/util/CloseableIterator.java Normal file
View File

@@ -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 {
}

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

@@ -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<T, V> implements Iterator<V> {
public class FunctionIterator<T, V> implements CloseableIterator<V> {
private final Iterator<T> iterator;
@@ -56,5 +59,11 @@ public 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();
}
}
}

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

@@ -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<Object> iterator = new Iterator<Object>() {
Iterator<Object> iterator = new CloseableIterator<Object>() {
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) {

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

@@ -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> {

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

@@ -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")