INT-651: Add Iterator Support for Splitter

JIRA: https://jira.spring.io/browse/INT-651

INT-651: Polishing according PR comments

INT-651: Polishing #2

Doc Polishing

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());
 	}

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<AbstractIntegrationMessageBuilder<?>> messageBuilders = new ArrayList<AbstractIntegrationMessageBuilder<?>>();
+
+		Iterator<Object> 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<Object> items = (Collection<Object>) 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<Object>) result).iterator();
+		}
+		else if (result instanceof Iterator<?>) {
+			sequenceSize = 0;
+			iterator = (Iterator<Object>) 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<Object, AbstractIntegrationMessageBuilder<?>>(iterator,
+				new Function<Object, AbstractIntegrationMessageBuilder<?>>() {
+					@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.
 	 */

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<T, V> implements Iterator<V> {
+
+	private final Iterator<T> iterator;
+
+	private final Function<? super T, ? extends V> function;
+
+	public FunctionIterator(Iterable<T> iterable, Function<? super T, ? extends V> function) {
+		this(iterable.iterator(), function);
+	}
+
+	public FunctionIterator(Iterator<T> newIterator, Function<? super T, ? extends V> 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();
+	}
+
+}
+

spring-integration-core/src/test/java/org/springframework/integration/splitter/SpelSplitterIntegrationTests-context.xml

@@ -17,6 +17,8 @@
 
 	<splitter input-channel="beanResolvingInput" expression="@testBean.split(payload)" output-channel="output"/>
 
+	<splitter input-channel="iteratorInput" ref="testBean" method="splitIterator" output-channel="output"/>
+
 	<beans:bean id="testBean" class="org.springframework.integration.splitter.SpelSplitterIntegrationTests$TestBean"/>
 
 </beans:beans>

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<String>("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<String> splitIterator(String s) {
+			return Arrays.asList(s.split(",")).iterator();
+		}
 	}
 
 }

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<String> split(String sentence) {
-			return Arrays.asList(sentence.split("\\s"));
+		public Iterator<String> split(String sentence) {
+			return Arrays.asList(sentence.split("\\s")).iterator();
 		}
 	}
 

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<String>("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<Message<?>> receivedMessages = replyChannel.clear();
+		Collections.sort(receivedMessages, new Comparator<Message<?>>() {
+
+			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<Message<?>> 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<String> annotatedMethod(String input) {
+			return new Iterator<String>() {
+
+				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<String> annotatedMethod(String input) {
+			return new Iterable<String>() {
+
+				public Iterator<String> iterator() {
+
+					return new Iterator<String>() {
+
+						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!");
+
+						}
+
+					};
+				}
+			};
+		}
+	}
+
+}