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 028b76283f..3878d0c424 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 @@ -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 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 index 00b4a63ced..f2bc40bece 100644 --- 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 @@ -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 extends Iterator, Closeable { +public interface CloseableIterator extends Iterator, AutoCloseable { + + @Override + void close(); // override throws Exception + } 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 d4ed6f0343..4b7f05abb4 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 @@ -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 implements CloseableIterator { + private final AutoCloseable closeable; + private final Iterator iterator; private final Function function; + /** + * Construct an instance with the provided iterable and function. + * @param iterable the iterable. + * @param function the function. + */ public FunctionIterator(Iterable iterable, Function 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 iterable, + Function 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 newIterator, Function 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 newIterator, + Function 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 implements CloseableIterator { } @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 + } } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/splitter/MethodInvokingSplitterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/splitter/MethodInvokingSplitterTests.java index 95b5ef551c..19dcee5f57 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/splitter/MethodInvokingSplitterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/splitter/MethodInvokingSplitterTests.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. @@ -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 message = new GenericMessage("foo.bar"); + GenericMessage 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 message = new GenericMessage("foo.bar"); + GenericMessage 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 message = new GenericMessage("foo.bar"); + GenericMessage 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 message = new GenericMessage("foo.bar"); + GenericMessage 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 message = new GenericMessage("foo.bar"); + GenericMessage 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 message = new GenericMessage("foo.bar"); + GenericMessage 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 message = MessageBuilder.withPayload("foo.bar") .setHeader("myHeader", "myValue") .build(); @@ -165,7 +171,7 @@ public class MethodInvokingSplitterTests { @Test public void splitStringToMessageArray() throws Exception { - GenericMessage message = new GenericMessage("foo.bar"); + GenericMessage 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 message = new GenericMessage("foo.bar"); + GenericMessage 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 message = new GenericMessage("foo.bar"); + public void splitStringToStringArrayConfiguredByMethodName() { + GenericMessage 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 message = new GenericMessage("foo.bar"); + public void splitStringToStringListConfiguredByMethodName() { + GenericMessage 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 message = new GenericMessage("foo.bar"); + public void splitMessageToStringArrayConfiguredByMethodName() { + GenericMessage 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 message = new GenericMessage("foo.bar"); + public void splitMessageToStringListConfiguredByMethodName() { + GenericMessage 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 message = new GenericMessage("foo.bar"); + public void splitMessageToMessageArrayConfiguredByMethodName() { + GenericMessage 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 message = new GenericMessage("foo.bar"); + public void splitMessageToMessageListConfiguredByMethodName() { + GenericMessage 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 message = new GenericMessage("foo.bar"); + public void splitStringToMessageArrayConfiguredByMethodName() { + GenericMessage 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 message = new GenericMessage("foo.bar"); + public void splitStringToMessageListConfiguredByMethodName() { + GenericMessage 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 split(List list) { return list; } } - GenericMessage> message = new GenericMessage>(Arrays.asList("foo", "bar")); + GenericMessage> 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 split(Stream stream) { + return stream; + } + + } + Stream 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) pf.getProxy(); + GenericMessage> message = new GenericMessage<>(stream); + MethodInvokingSplitter splitter = new MethodInvokingSplitter(new StreamSplitter(), "split"); + QueueChannel replyChannel = new QueueChannel(); + splitter.setOutputChannel(replyChannel); + splitter.handleMessage(message); + List> 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 message = new GenericMessage("foo.bar"); + GenericMessage 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 message = new GenericMessage("foo.bar"); + GenericMessage 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 message = new GenericMessage("foo.bar"); + GenericMessage 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 message = new GenericMessage("foo.bar"); + GenericMessage 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> messageToMessageList(Message input) { String[] strings = input.getPayload().toString().split("\\."); - List> messages = new ArrayList>(); + List> messages = new ArrayList<>(); for (String s : strings) { messages.add(new GenericMessage(s)); } @@ -533,9 +577,9 @@ public class MethodInvokingSplitterTests { public List> stringToMessageList(String input) { String[] strings = input.split("\\."); - List> messages = new ArrayList>(); + List> messages = new ArrayList<>(); for (String s : strings) { - messages.add(new GenericMessage(s)); + messages.add(new GenericMessage<>(s)); } return messages; } @@ -546,7 +590,7 @@ public class MethodInvokingSplitterTests { public List splitPayloadAndHeader(String payload, @Header("testHeader") String header) { String regex = "\\."; - List results = new ArrayList(); + List results = new ArrayList<>(); Collections.addAll(results, payload.split(regex)); Collections.addAll(results, header.split(regex)); return results; @@ -614,4 +658,5 @@ public class MethodInvokingSplitterTests { } } + } diff --git a/spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.java b/spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.java index 4a6f06f6aa..cb346de8a2 100644 --- a/spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.java +++ b/spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.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. @@ -329,7 +329,7 @@ public class XPathMessageSplitter extends AbstractMessageSplitter { TransformFunctionIterator(Iterator delegate, Function function) { - super(delegate, function); + super(null, delegate, function); this.delegate = delegate; }