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

INT-4506: Splitter: close stream and Closeables

Browse Source 
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**
This commit is contained in:
Gary Russell
2018-07-25 10:45:50 -04:00
committed by Artem Bilan
parent 0d0605be78
commit bebde260db
5 changed files with 153 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

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

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

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

117
spring-integration-core/src/test/java/org/springframework/integration/splitter/MethodInvokingSplitterTests.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.
@@ -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 {
GenericMessage<String> message = new GenericMessage<String>("foo.bar");
public void splitStringToStringArrayConfiguredByMethodName() {
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 {
GenericMessage<String> message = new GenericMessage<String>("foo.bar");
public void splitStringToStringListConfiguredByMethodName() {
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 {
GenericMessage<String> message = new GenericMessage<String>("foo.bar");
public void splitMessageToStringArrayConfiguredByMethodName() {
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 {
GenericMessage<String> message = new GenericMessage<String>("foo.bar");
public void splitMessageToStringListConfiguredByMethodName() {
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 {
GenericMessage<String> message = new GenericMessage<String>("foo.bar");
public void splitMessageToMessageArrayConfiguredByMethodName() {
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 {
GenericMessage<String> message = new GenericMessage<String>("foo.bar");
public void splitMessageToMessageListConfiguredByMethodName() {
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 {
GenericMessage<String> message = new GenericMessage<String>("foo.bar");
public void splitStringToMessageArrayConfiguredByMethodName() {
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 {
GenericMessage<String> message = new GenericMessage<String>("foo.bar");
public void splitStringToMessageListConfiguredByMethodName() {
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 {
}
}
}

4
spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.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.
@@ -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;
}