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

Author: Artem Bilan

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java

@@ -200,7 +200,7 @@ else if (result != null) {
 		}
 	}
 
-	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,73 +16,94 @@
 
 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 {
 
 	private boolean applySequence = true;
 
 	/**
 	 * 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();
+		}
+
+		if (!iterator.hasNext()) {
+			return null;
 		}
-		return messageBuilders;
+
+		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 @@ private AbstractIntegrationMessageBuilder createBuilder(Object item, MessageHead
 		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 String getComponentType() {
 	 * 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 void beanResolving() {
 		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 List<Integer> getNumbers() {
 		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 java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Iterator;
 import java.util.List;
 
 import org.junit.Before;
@@ -95,8 +96,8 @@ public void deliveredWords(String string) {
 	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();
 		}
 	}