diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 04dac0644f..dc37a28ac4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -200,7 +200,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa } } - private void produceReply(Object reply, MessageHeaders requestHeaders) { + protected void produceReply(Object reply, MessageHeaders requestHeaders) { Message replyMessage = this.createReplyMessage(reply, requestHeaders); this.sendReplyMessage(replyMessage, requestHeaders.getReplyChannel()); } 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 8083b70cba..3a0e4b4e28 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,22 +16,26 @@ package org.springframework.integration.splitter; -import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; -import java.util.List; +import java.util.Collections; +import java.util.Iterator; +import java.util.concurrent.atomic.AtomicInteger; + +import reactor.function.Function; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.integration.util.FunctionIterator; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; -import org.springframework.util.CollectionUtils; -import org.springframework.util.ObjectUtils; /** * Base class for Message-splitting handlers. * * @author Mark Fisher * @author Dave Syer + * @author Artem Bilan */ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler { @@ -39,50 +43,67 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess /** * Set the applySequence flag to the specified value. Defaults to true. - * * @param applySequence true to apply sequence information. */ public void setApplySequence(boolean applySequence) { this.applySequence = applySequence; } - @SuppressWarnings("rawtypes") @Override + @SuppressWarnings("unchecked") protected final Object handleRequestMessage(Message message) { Object result = this.splitMessage(message); - // return null if 'null', empty Collection or empty Array - if (result == null || (result instanceof Collection && CollectionUtils.isEmpty((Collection) result)) - || (result.getClass().isArray() && ObjectUtils.isEmpty((Object[]) result))) { + // return null if 'null' + if (result == null) { return null; } - MessageHeaders headers = message.getHeaders(); - Object correlationId = headers.getId(); - List> messageBuilders = new ArrayList>(); + + Iterator iterator; + final int sequenceSize; if (result instanceof Collection) { - Collection items = (Collection) result; - int sequenceNumber = 0; - int sequenceSize = items.size(); - for (Object item : items) { - messageBuilders.add(this.createBuilder(item, headers, correlationId, ++sequenceNumber, sequenceSize)); - } + Collection items = (Collection) result; + sequenceSize = items.size(); + iterator = items.iterator(); } else if (result.getClass().isArray()) { Object[] items = (Object[]) result; - int sequenceNumber = 0; - int sequenceSize = items.length; - for (Object item : items) { - messageBuilders.add(this.createBuilder(item, headers, correlationId, ++sequenceNumber, sequenceSize)); - } + sequenceSize = items.length; + iterator = Arrays.asList(items).iterator(); + } + else if (result instanceof Iterable) { + sequenceSize = 0; + iterator = ((Iterable) result).iterator(); + } + else if (result instanceof Iterator) { + sequenceSize = 0; + iterator = (Iterator) result; } else { - messageBuilders.add(this.createBuilder(result, headers, correlationId, 1, 1)); + sequenceSize = 1; + iterator = Collections.singleton(result).iterator(); } - return messageBuilders; + + if (!iterator.hasNext()) { + return null; + } + + final MessageHeaders headers = message.getHeaders(); + final Object correlationId = headers.getId(); + final AtomicInteger sequenceNumber = new AtomicInteger(1); + + return new FunctionIterator>(iterator, + new Function>() { + @Override + public AbstractIntegrationMessageBuilder apply(Object object) { + return createBuilder(object, headers, correlationId, sequenceNumber.getAndIncrement(), + sequenceSize); + } + }); } @SuppressWarnings( { "unchecked", "rawtypes" }) - private AbstractIntegrationMessageBuilder createBuilder(Object item, MessageHeaders headers, Object correlationId, int sequenceNumber, - int sequenceSize) { + private AbstractIntegrationMessageBuilder createBuilder(Object item, MessageHeaders headers, Object correlationId, + int sequenceNumber, int sequenceSize) { AbstractIntegrationMessageBuilder builder; if (item instanceof Message) { builder = this.getMessageBuilderFactory().fromMessage((Message) item); @@ -97,6 +118,15 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess return builder; } + @Override + protected void produceReply(Object result, MessageHeaders requestHeaders) { + Iterator iterator = (Iterator) result; + while (iterator.hasNext()) { + super.produceReply(iterator.next(), requestHeaders); + + } + } + @Override public String getComponentType() { return "splitter"; @@ -107,7 +137,6 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess * Array. The individual elements may be Messages, but it is not necessary. If the elements are not Messages, each * will be provided as the payload of a Message. It is also acceptable to return a single Object or Message. In that * case, a single reply Message will be produced. - * * @param message The message. * @return The result of splitting the message. */ 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 new file mode 100644 index 0000000000..749e0de7cf --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/FunctionIterator.java @@ -0,0 +1,65 @@ +/* + * Copyright 2014 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.util.Iterator; +import java.util.NoSuchElementException; + +import reactor.function.Function; + +/** + * 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 + * @since 4.1 + */ +public final class FunctionIterator implements Iterator { + + private final Iterator iterator; + + private final Function function; + + public FunctionIterator(Iterable iterable, Function function) { + this(iterable.iterator(), function); + } + + public FunctionIterator(Iterator newIterator, Function function) { + 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(); + } + + @Override + public V next() { + if (this.hasNext()) { + return this.function.apply(this.iterator.next()); + } + throw new NoSuchElementException(); + } + +} + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/splitter/SpelSplitterIntegrationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/splitter/SpelSplitterIntegrationTests-context.xml index 7ad5a57a6a..3753adfbad 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/splitter/SpelSplitterIntegrationTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/splitter/SpelSplitterIntegrationTests-context.xml @@ -17,6 +17,8 @@ + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/splitter/SpelSplitterIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/splitter/SpelSplitterIntegrationTests.java index 32d901fc24..c7f5cad041 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/splitter/SpelSplitterIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/splitter/SpelSplitterIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2014 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,26 +16,29 @@ package org.springframework.integration.splitter; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; +import static org.junit.Assert.*; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Iterator; import java.util.List; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Mark Fisher + * @author Artem Bilan */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @@ -50,6 +53,9 @@ public class SpelSplitterIntegrationTests { @Autowired private MessageChannel beanResolvingInput; + @Autowired + private MessageChannel iteratorInput; + @Autowired private PollableChannel output; @@ -99,6 +105,28 @@ public class SpelSplitterIntegrationTests { assertNull(output.receive(0)); } + @Test + public void iteratorSplitter() { + this.iteratorInput.send(new GenericMessage("a,b,c,d")); + Message a = output.receive(0); + Message b = output.receive(0); + Message c = output.receive(0); + Message d = output.receive(0); + assertEquals("a", a.getPayload()); + assertEquals(new Integer(1), new IntegrationMessageHeaderAccessor(a).getSequenceNumber()); + assertEquals(new Integer(0), new IntegrationMessageHeaderAccessor(a).getSequenceSize()); + assertEquals("b", b.getPayload()); + assertEquals(new Integer(2), new IntegrationMessageHeaderAccessor(b).getSequenceNumber()); + assertEquals(new Integer(0), new IntegrationMessageHeaderAccessor(b).getSequenceSize()); + assertEquals("c", c.getPayload()); + assertEquals(new Integer(3), new IntegrationMessageHeaderAccessor(c).getSequenceNumber()); + assertEquals(new Integer(0), new IntegrationMessageHeaderAccessor(c).getSequenceSize()); + assertEquals("d", d.getPayload()); + assertEquals(new Integer(4), new IntegrationMessageHeaderAccessor(d).getSequenceNumber()); + assertEquals(new Integer(0), new IntegrationMessageHeaderAccessor(d).getSequenceSize()); + assertNull(output.receive(0)); + } + static class TestBean { @@ -117,6 +145,10 @@ public class SpelSplitterIntegrationTests { public String[] split(String s) { return s.split(","); } + + public Iterator splitIterator(String s) { + return Arrays.asList(s.split(",")).iterator(); + } } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/splitter/SplitterIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/splitter/SplitterIntegrationTests.java index e2161f52ca..81a86bace7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/splitter/SplitterIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/splitter/SplitterIntegrationTests.java @@ -21,6 +21,7 @@ import static org.junit.Assert.assertTrue; import java.util.ArrayList; import java.util.Arrays; +import java.util.Iterator; import java.util.List; import org.junit.Before; @@ -95,8 +96,8 @@ public class SplitterIntegrationTests { public static class TestSplitter { @Splitter(inputChannel = "inAnnotated", outputChannel = "out") - public List split(String sentence) { - return Arrays.asList(sentence.split("\\s")); + public Iterator split(String sentence) { + return Arrays.asList(sentence.split("\\s")).iterator(); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/splitter/StreamingSplitterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/splitter/StreamingSplitterTests.java new file mode 100644 index 0000000000..576662876b --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/splitter/StreamingSplitterTests.java @@ -0,0 +1,260 @@ +/* + * Copyright 2002-2011 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.splitter; + +import static org.hamcrest.Matchers.*; +import static org.junit.Assert.*; + +import java.util.Collections; +import java.util.Comparator; +import java.util.Iterator; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Before; +import org.junit.Test; + +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.MessageRejectedException; +import org.springframework.integration.annotation.Splitter; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.GenericMessage; + +/** + * @author Alex Peters + * @author Artem Bilan + * @since 4.1 + */ +public class StreamingSplitterTests { + + private Message message; + + @Before + public void setUp() { + message = new GenericMessage("foo.bar"); + } + + + @Test + public void splitToIterator_sequenceSizeInLastMessageHeader() + throws Exception { + int messageQuantity = 5; + MethodInvokingSplitter splitter = new MethodInvokingSplitter(new IteratorTestBean( + messageQuantity)); + QueueChannel replyChannel = new QueueChannel(); + splitter.setOutputChannel(replyChannel); + + splitter.handleMessage(message); + List> receivedMessages = replyChannel.clear(); + Collections.sort(receivedMessages, new Comparator>() { + + public int compare(Message o1, Message o2) { + return o1.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, Integer.class) + .compareTo(o2.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, Integer.class)); + } + + }); + assertThat(receivedMessages.get(4) + .getHeaders() + .get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, Integer.class), + is(messageQuantity)); + } + + @Test + public void splitToIterator_sourceMessageHeadersIncluded() throws Exception { + String anyHeaderKey = "anyProperty1"; + String anyHeaderValue = "anyValue1"; + message = MessageBuilder.fromMessage(message) + .setHeader(anyHeaderKey, anyHeaderValue) + .build(); + int messageQuantity = 5; + MethodInvokingSplitter splitter = new MethodInvokingSplitter(new IteratorTestBean( + messageQuantity)); + QueueChannel replyChannel = new QueueChannel(); + splitter.setOutputChannel(replyChannel); + splitter.handleMessage(message); + List> receivedMessages = replyChannel.clear(); + assertThat(receivedMessages.size(), is(messageQuantity)); + for (Message reveivedMessage : receivedMessages) { + MessageHeaders headers = reveivedMessage.getHeaders(); + assertTrue("Unexpected result with: " + headers, headers.containsKey(anyHeaderKey)); + assertThat("Unexpected result with: " + headers, + headers.get(anyHeaderKey, String.class), + is(anyHeaderValue)); + assertThat("Unexpected result with: " + headers, + headers.get(IntegrationMessageHeaderAccessor.CORRELATION_ID, UUID.class), + is(message.getHeaders().getId())); + } + } + + @Test + public void splitToIterator_allMessagesSent() throws Exception { + int messageQuantity = 5; + MethodInvokingSplitter splitter = new MethodInvokingSplitter(new IteratorTestBean( + messageQuantity)); + QueueChannel replyChannel = new QueueChannel(); + splitter.setOutputChannel(replyChannel); + + splitter.handleMessage(message); + assertThat(replyChannel.getQueueSize(), is(messageQuantity)); + } + + @Test + public void splitToIterable_allMessagesSent() throws Exception { + int messageQuantity = 5; + MethodInvokingSplitter splitter = new MethodInvokingSplitter(new IterableTestBean( + messageQuantity)); + QueueChannel replyChannel = new QueueChannel(); + splitter.setOutputChannel(replyChannel); + + splitter.handleMessage(message); + assertThat(replyChannel.getQueueSize(), is(messageQuantity)); + } + + @Test + public void splitToIterator_allMessagesContainSequenceNumber() + throws Exception { + final int messageQuantity = 5; + MethodInvokingSplitter splitter = new MethodInvokingSplitter(new IteratorTestBean( + messageQuantity)); + DirectChannel replyChannel = new DirectChannel(); + splitter.setOutputChannel(replyChannel); + + new EventDrivenConsumer(replyChannel, new MessageHandler() { + + public void handleMessage(Message message) throws MessagingException { + assertThat("Failure with msg: " + message, + message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, Integer.class), + is(Integer.valueOf((String) message.getPayload()))); + } + }).start(); + splitter.handleMessage(message); + } + + @Test + public void splitWithMassiveReplyMessages_allMessagesSent() + throws Exception { + final int messageQuantity = 100000; + MethodInvokingSplitter splitter = new MethodInvokingSplitter(new IteratorTestBean( + messageQuantity)); + DirectChannel replyChannel = new DirectChannel(); + splitter.setOutputChannel(replyChannel); + + final AtomicInteger receivedMessageCounter = new AtomicInteger(0); + new EventDrivenConsumer(replyChannel, new MessageHandler() { + + public void handleMessage(Message message) + throws MessageRejectedException, MessageHandlingException, + MessageDeliveryException { + assertThat("Failure with msg: " + message, + message.getPayload(), + is(notNullValue())); + receivedMessageCounter.incrementAndGet(); + + } + }).start(); + + splitter.handleMessage(message); + assertThat(receivedMessageCounter.get(), is(messageQuantity)); + } + + static class IteratorTestBean { + + final int max; + + AtomicInteger counter = new AtomicInteger(0); + + public IteratorTestBean(int max) { + this.max = max; + } + + @Splitter + public Iterator annotatedMethod(String input) { + return new Iterator() { + + public boolean hasNext() { + return counter.get() < max; + } + + public String next() { + if (!hasNext()) { + throw new IllegalStateException("Last element reached!"); + } + return String.valueOf(counter.incrementAndGet()); + } + + public void remove() { + throw new AssertionError("not implemented!"); + + } + + }; + } + } + + static class IterableTestBean { + + final int max; + + AtomicInteger counter = new AtomicInteger(0); + + public IterableTestBean(int max) { + this.max = max; + } + + @Splitter + public Iterable annotatedMethod(String input) { + return new Iterable() { + + public Iterator iterator() { + + return new Iterator() { + + public boolean hasNext() { + return counter.get() < max; + } + + public String next() { + if (!hasNext()) { + throw new IllegalStateException( + "Last element reached!"); + } + return String.valueOf(counter.incrementAndGet()); + } + + public void remove() { + throw new AssertionError("not implemented!"); + + } + + }; + } + }; + } + } + +} diff --git a/src/reference/docbook/splitter.xml b/src/reference/docbook/splitter.xml index 11aa0f95ba..4e28a125cf 100644 --- a/src/reference/docbook/splitter.xml +++ b/src/reference/docbook/splitter.xml @@ -44,8 +44,9 @@ - a Collection (or subclass thereof) or an array of - Message objects - + A Collection or an array of Messages, + or an Iterable (or Iterator) + that iterates over Messages - in this case the messages will be sent as such (after the CORRELATION_ID, SEQUENCE_SIZE and SEQUENCE_NUMBER are populated). Using this approach gives more control to the developer, for example @@ -54,8 +55,9 @@ - a Collection (or subclass thereof) or an array of - non-Message objects - works like the prior case, except that each collection + A Collection or an array of non-Message objects, + or an Iterable (or Iterator) + that iterates over non-Message objects - works like the prior case, except that each collection element will be used as a Message payload. Using this approach allows developers to focus on the domain objects without having to consider the Messaging system and produces code that is easier to test. @@ -76,6 +78,27 @@ In the latter case, the splitter will receive the payload of the incoming message. Since this decouples the code from the Spring Integration API and will typically be easier to test, it is the recommended approach. + + Splitter and Iterators + + + Starting with version 4.1, the AbstractMessageSplitter + supports the Iterator type for the value to split. + Note, in the case of an Iterator + (or Iterable), we don't have access to the number of underlying items and the + SEQUENCE_SIZE header is set to 0. This means that the default + SequenceSizeReleaseStrategy of an <aggregator> won't work and the + group for the CORRELATION_ID from the splitter won't be released; it will remain + as incomplete. In this case you should use an appropriate custom + ReleaseStrategy or rely on send-partial-result-on-expiry + together with group-timeout or a MessageGroupStoreReaper. + + + An Iterator object is useful to avoid the need for building an entire + collection in the memory before splitting. For example, when underlying items are populated from some external system + (e.g. DataBase or FTP MGET) using iterations or streams. + + diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 521c2f1249..8cec88edfd 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -379,5 +379,13 @@ For more information, see . +
+ Splitter and Iterator + + Splitter components now support an Iterator as the result object + for producing output messages. + See for more information. + +