Added AbstractConsumerFactoryBean and SplitterFactoryBean to support different consumer types that can only be initialized after the "ref" Object's type is known (for example, see INT-397).

org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractConsumerFactoryBean.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2002-2008 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.config;
+
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
+import org.springframework.integration.message.MessageConsumer;
+import org.springframework.util.Assert;
+
+/**
+ * Base class for FactoryBeans that create MessageConsumer instances.
+ * 
+ * @author Mark Fisher
+ */
+public abstract class AbstractConsumerFactoryBean implements FactoryBean {
+
+	private volatile MessageConsumer consumer;
+
+	private volatile Object targetObject;
+
+	private volatile String targetMethodName;
+
+	private volatile MessageChannel outputChannel;
+
+	private volatile boolean initialized;
+
+	private final Object initializationMonitor = new Object();
+
+
+	public void setTargetObject(Object targetObject) {
+		this.targetObject = targetObject;
+	}
+
+	public void setTargetMethodName(String targetMethodName) {
+		this.targetMethodName = targetMethodName;
+	}
+
+	public void setOutputChannel(MessageChannel outputChannel) {
+		this.outputChannel = outputChannel;
+	}
+
+	public Object getObject() throws Exception {
+		if (this.consumer == null) {
+			this.initializeConsumer();
+			Assert.notNull(this.consumer, "failed to create MessageConsumer");
+			if (this.outputChannel != null
+					&& this.consumer instanceof AbstractReplyProducingMessageConsumer) {
+				((AbstractReplyProducingMessageConsumer) this.consumer).setOutputChannel(this.outputChannel);
+			}
+		}
+		return this.consumer;
+	}
+
+	public Class<?> getObjectType() {
+		if (this.consumer != null) {
+			return this.consumer.getClass();
+		}
+		return MessageConsumer.class;
+	}
+
+	public boolean isSingleton() {
+		return true;
+	}
+
+	private void initializeConsumer() {
+		synchronized (this.initializationMonitor) {
+			if (this.initialized) {
+				return;
+			}
+			this.consumer = this.createConsumer(this.targetObject, this.targetMethodName);
+			this.initialized = true;
+		}
+	}
+
+	/**
+	 * Subclasses must implement this method to create the MessageConsumer.
+	 */
+	protected abstract MessageConsumer createConsumer(Object targetObject, String targetMethodName);
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2002-2008 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.config;
+
+import org.springframework.integration.message.MessageConsumer;
+import org.springframework.integration.splitter.AbstractMessageSplitter;
+import org.springframework.integration.splitter.DefaultMessageSplitter;
+import org.springframework.integration.splitter.MethodInvokingSplitter;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * Factory bean for creating a Message Splitter.
+ * 
+ * @author Mark Fisher
+ */
+public class SplitterFactoryBean extends AbstractConsumerFactoryBean {
+
+	@Override
+	protected MessageConsumer createConsumer(Object targetObject, String targetMethodName) {
+		if (targetObject == null) {
+			Assert.isTrue(!StringUtils.hasText(targetMethodName),
+					"'method' should only be provided when 'ref' is also provided");
+			return new DefaultMessageSplitter();
+		}
+		if (targetObject instanceof AbstractMessageSplitter) {
+			return (AbstractMessageSplitter) targetObject;
+		}
+		return (StringUtils.hasText(targetMethodName))
+				? new MethodInvokingSplitter(targetObject, targetMethodName)
+				: new MethodInvokingSplitter(targetObject);
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/config/SplitterParser.java

@@ -20,8 +20,6 @@ import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
-import org.springframework.integration.splitter.DefaultMessageSplitter;
-import org.springframework.integration.splitter.MethodInvokingSplitter;
 import org.springframework.util.StringUtils;
 
 /**
@@ -33,17 +31,16 @@ public class SplitterParser extends AbstractConsumerEndpointParser {
 
 	@Override
 	protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(SplitterFactoryBean.class);
 		if (element.hasAttribute(REF_ATTRIBUTE)) {
 			String ref = element.getAttribute(REF_ATTRIBUTE);
-			BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingSplitter.class);
-			builder.addConstructorArgReference(ref);
+			builder.addPropertyReference("targetObject", ref);
 			if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
 				String method = element.getAttribute(METHOD_ATTRIBUTE);
-				builder.addConstructorArgValue(method);
+				builder.addPropertyValue("targetMethodName", method);
 			}
-			return builder;
 		}
-		return BeanDefinitionBuilder.genericBeanDefinition(DefaultMessageSplitter.class);
+		return builder;
 	}
 
 }

org.springframework.integration/src/test/java/org/springframework/integration/router/config/SplitterParserTests.java

@@ -33,22 +33,60 @@ import org.springframework.integration.message.StringMessage;
 public class SplitterParserTests {
 
 	@Test
-	public void testSplitter() {
+	public void splitterAdapterWithRefAndMethod() {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"splitterParserTests.xml", this.getClass());
 		context.start();
-		MessageChannel channel1 = (MessageChannel) context.getBean("channel1");
-		PollableChannel channel2 = (PollableChannel) context.getBean("channel2");
-		channel1.send(new StringMessage("this.is.a.test"));
-		Message<?> result1 = channel2.receive(1000);
+		MessageChannel input = (MessageChannel) context.getBean("splitterAdapterWithRefAndMethodInput");
+		PollableChannel output = (PollableChannel) context.getBean("output");
+		input.send(new StringMessage("this.is.a.test"));
+		Message<?> result1 = output.receive(1000);
 		assertEquals("this", result1.getPayload());
-		Message<?> result2 = channel2.receive(1000);
+		Message<?> result2 = output.receive(1000);
 		assertEquals("is", result2.getPayload());
-		Message<?> result3 = channel2.receive(1000);
+		Message<?> result3 = output.receive(1000);
 		assertEquals("a", result3.getPayload());
-		Message<?> result4 = channel2.receive(1000);
+		Message<?> result4 = output.receive(1000);
 		assertEquals("test", result4.getPayload());
-		assertNull(channel2.receive(0));
+		assertNull(output.receive(0));
+	}
+
+	@Test
+	public void splitterAdapterWithRefOnly() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"splitterParserTests.xml", this.getClass());
+		context.start();
+		MessageChannel input = (MessageChannel) context.getBean("splitterAdapterWithRefOnlyInput");
+		PollableChannel output = (PollableChannel) context.getBean("output");
+		input.send(new StringMessage("this.is.a.test"));
+		Message<?> result1 = output.receive(1000);
+		assertEquals("this", result1.getPayload());
+		Message<?> result2 = output.receive(1000);
+		assertEquals("is", result2.getPayload());
+		Message<?> result3 = output.receive(1000);
+		assertEquals("a", result3.getPayload());
+		Message<?> result4 = output.receive(1000);
+		assertEquals("test", result4.getPayload());
+		assertNull(output.receive(0));
+	}
+
+	@Test
+	public void splitterImplementation() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"splitterParserTests.xml", this.getClass());
+		context.start();
+		MessageChannel input = (MessageChannel) context.getBean("splitterImplementationInput");
+		PollableChannel output = (PollableChannel) context.getBean("output");
+		input.send(new StringMessage("this.is.a.test"));
+		Message<?> result1 = output.receive(1000);
+		assertEquals("this", result1.getPayload());
+		Message<?> result2 = output.receive(1000);
+		assertEquals("is", result2.getPayload());
+		Message<?> result3 = output.receive(1000);
+		assertEquals("a", result3.getPayload());
+		Message<?> result4 = output.receive(1000);
+		assertEquals("test", result4.getPayload());
+		assertNull(output.receive(0));
 	}
 
 }

org.springframework.integration/src/test/java/org/springframework/integration/router/config/TestSplitterBean.java

@@ -19,7 +19,7 @@ package org.springframework.integration.router.config;
 /**
  * @author Mark Fisher
  */
-public class TestSplitter {
+public class TestSplitterBean {
 
 	public String[] split(String input) {
 		return input.split("\\.");

org.springframework.integration/src/test/java/org/springframework/integration/router/config/TestSplitterImpl.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2002-2008 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.router.config;
+
+import org.springframework.integration.message.Message;
+import org.springframework.integration.splitter.AbstractMessageSplitter;
+
+/**
+ * @author Mark Fisher
+ */
+public class TestSplitterImpl extends AbstractMessageSplitter {
+
+
+	@Override
+	protected Object splitMessage(Message<?> message) {
+		return message.getPayload().toString().split("\\.");
+	}
+
+}

org.springframework.integration/src/test/java/org/springframework/integration/router/config/splitterParserTests.xml

@@ -9,15 +9,28 @@
 
 	<message-bus/>
 
-	<channel id="channel1"/>
-
-	<channel id="channel2">
-		<queue capacity="5"/>
+	<channel id="output">
+		<queue capacity="10"/>
 	</channel>
 
-	<splitter id="splitter" ref="pojo" method="split"
-	          input-channel="channel1" output-channel="channel2"/>
+	<splitter id="splitterAdapterWithRefAndMethod"
+			  ref="splitterBean"
+			  method="split"
+	          input-channel="splitterAdapterWithRefAndMethodInput"
+	          output-channel="output"/>
 
-	<beans:bean id="pojo" class="org.springframework.integration.router.config.TestSplitter"/>
+	<splitter id="splitterAdapterWithRefOnly"
+			  ref="splitterBean"
+	          input-channel="splitterAdapterWithRefOnlyInput"
+	          output-channel="output"/>
+
+	<splitter id="splitterImplementation"
+			  ref="splitterImpl"
+	          input-channel="splitterImplementationInput"
+	          output-channel="output"/>
+
+	<beans:bean id="splitterBean" class="org.springframework.integration.router.config.TestSplitterBean"/>
+
+	<beans:bean id="splitterImpl" class="org.springframework.integration.router.config.TestSplitterImpl"/>
 
 </beans:beans>