diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/AbstractMessageHandlingSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/AbstractMessageHandlingSourceAdapter.java
new file mode 100644
index 0000000000..e180a45504
--- /dev/null
+++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/AbstractMessageHandlingSourceAdapter.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright 2002-2007 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.adapter;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.MessagingConfigurationException;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.RequestReplyTemplate;
+import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.message.Message;
+
+/**
+ * Abstract base class for source adapters that handle request Messages.
+ *
+ * @author Mark Fisher
+ */
+public abstract class AbstractMessageHandlingSourceAdapter implements SourceAdapter, MessageHandler, InitializingBean {
+
+ private final Log logger = LogFactory.getLog(this.getClass());
+
+ private volatile MessageChannel channel;
+
+ private volatile RequestReplyTemplate requestReplyTemplate;
+
+ private volatile boolean expectReply = true;
+
+ private volatile long sendTimeout = -1;
+
+ private volatile long receiveTimeout = -1;
+
+ protected final Object lifecycleMonitor = new Object();
+
+ private volatile boolean initialized;
+
+
+ public void setChannel(MessageChannel channel) {
+ this.channel = channel;
+ }
+
+ protected MessageChannel getChannel() {
+ return this.channel;
+ }
+
+ /**
+ * Specify whether the handle method should be expected to return a reply.
+ * The default is 'true'.
+ */
+ public void setExpectReply(boolean expectReply) {
+ this.expectReply = expectReply;
+ }
+
+ public void setSendTimeout(long sendTimeout) {
+ this.sendTimeout = sendTimeout;
+ }
+
+ public void setReceiveTimeout(long receiveTimeout) {
+ this.receiveTimeout = receiveTimeout;
+ }
+
+ public final void afterPropertiesSet() throws Exception {
+ synchronized (this.lifecycleMonitor) {
+ if (this.initialized) {
+ return;
+ }
+ if (this.requestReplyTemplate == null) {
+ this.requestReplyTemplate = this.createRequestReplyTemplate();
+ }
+ }
+ this.initialize();
+ this.initialized = true;
+ }
+
+ /**
+ * Subclasses may override this method for initialization.
+ */
+ protected void initialize() throws Exception {
+ }
+
+ private RequestReplyTemplate createRequestReplyTemplate() {
+ RequestReplyTemplate template = new RequestReplyTemplate(this.channel);
+ template.setDefaultSendTimeout(this.sendTimeout);
+ template.setDefaultReceiveTimeout(this.receiveTimeout);
+ return template;
+ }
+
+ public final Message handle(Message message) {
+ if (!this.initialized) {
+ try {
+ this.afterPropertiesSet();
+ }
+ catch (Exception e) {
+ throw new MessagingConfigurationException("unable to initialize " + this.getClass().getName(), e);
+ }
+ }
+ if (!this.expectReply) {
+ if (!this.channel.send(message, this.sendTimeout) && logger.isWarnEnabled()) {
+ logger.warn("failed to send message to channel '" + this.channel +
+ "' within timeout of " + this.sendTimeout + " milliseconds");
+ }
+ return null;
+ }
+ return this.requestReplyTemplate.request(message);
+ }
+
+}
diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java
index 8427c37128..c348d4bf2c 100644
--- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java
+++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java
@@ -18,16 +18,9 @@ package org.springframework.integration.adapter.rmi;
import java.rmi.RemoteException;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.MessagingConfigurationException;
-import org.springframework.integration.adapter.SourceAdapter;
-import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.channel.RequestReplyTemplate;
+import org.springframework.integration.adapter.AbstractMessageHandlingSourceAdapter;
import org.springframework.integration.handler.MessageHandler;
-import org.springframework.integration.message.Message;
import org.springframework.remoting.rmi.RmiServiceExporter;
/**
@@ -35,52 +28,16 @@ import org.springframework.remoting.rmi.RmiServiceExporter;
*
* @author Mark Fisher
*/
-public class RmiSourceAdapter implements SourceAdapter, MessageHandler, InitializingBean {
+public class RmiSourceAdapter extends AbstractMessageHandlingSourceAdapter {
public static final String SERVICE_NAME_PREFIX = "internal.rmiSourceAdapter.";
- private final Log logger = LogFactory.getLog(this.getClass());
-
- private volatile MessageChannel channel;
-
- private volatile RequestReplyTemplate requestReplyTemplate;
-
- private volatile boolean expectReply = true;
-
- private volatile long sendTimeout = -1;
-
- private volatile long receiveTimeout = -1;
-
-
- public void setChannel(MessageChannel channel) {
- this.channel = channel;
- }
-
- /**
- * Specify whether the handle method should be expected to return a reply.
- * The default is 'true'.
- */
- public void setExpectReply(boolean expectReply) {
- this.expectReply = expectReply;
- }
-
- public void setSendTimeout(long sendTimeout) {
- this.sendTimeout = sendTimeout;
- }
-
- public void setReceiveTimeout(long receiveTimeout) {
- this.receiveTimeout = receiveTimeout;
- }
-
- public void afterPropertiesSet() throws RemoteException {
- String channelName = this.channel.getName();
+ public void initialize() throws RemoteException {
+ String channelName = this.getChannel().getName();
if (channelName == null) {
throw new MessagingConfigurationException("RmiSourceAdapter's MessageChannel must have a 'name'");
}
- this.requestReplyTemplate = new RequestReplyTemplate(this.channel);
- this.requestReplyTemplate.setDefaultSendTimeout(this.sendTimeout);
- this.requestReplyTemplate.setDefaultReceiveTimeout(this.receiveTimeout);
RmiServiceExporter exporter = new RmiServiceExporter();
exporter.setService(this);
exporter.setServiceInterface(MessageHandler.class);
@@ -88,23 +45,4 @@ public class RmiSourceAdapter implements SourceAdapter, MessageHandler, Initiali
exporter.afterPropertiesSet();
}
- public Message handle(Message message) {
- if (this.requestReplyTemplate == null) {
- try {
- this.afterPropertiesSet();
- }
- catch (RemoteException e) {
- throw new MessagingConfigurationException("unable to initialize RmiSourceAdapter", e);
- }
- }
- if (!this.expectReply) {
- if (!this.channel.send(message, this.sendTimeout) && logger.isWarnEnabled()) {
- logger.warn("failed to send message to channel '" + channel +
- "' within timeout of " + this.sendTimeout + " milliseconds");
- }
- return null;
- }
- return this.requestReplyTemplate.request(message);
- }
-
}
diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java
index 0eca33abee..8e8a299783 100644
--- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java
+++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java
@@ -19,8 +19,6 @@ package org.springframework.integration.adapter.rmi.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
-import java.rmi.RemoteException;
-
import org.junit.Before;
import org.junit.Test;
@@ -41,7 +39,7 @@ public class RmiTargetAdapterParserTests {
@Before
- public void exportRemoteHandler() throws RemoteException {
+ public void exportRemoteHandler() throws Exception {
testChannel.setBeanName("testChannel");
RmiSourceAdapter sourceAdapter = new RmiSourceAdapter();
sourceAdapter.setChannel(testChannel);
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java
index 54da49d6e5..2919832d79 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java
@@ -26,7 +26,6 @@ import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
-import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHeader;
import org.springframework.integration.util.SimpleMethodInvoker;