From 9f66c37b4cb7b1ff5aab52b61919aafc461296f7 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 7 Mar 2008 23:48:13 +0000 Subject: [PATCH] Factored out common remoting-based source adapter behavior into AbstractMessageHandlingSourceAdapter. --- .../AbstractMessageHandlingSourceAdapter.java | 122 ++++++++++++++++++ .../adapter/rmi/RmiSourceAdapter.java | 70 +--------- .../config/RmiTargetAdapterParserTests.java | 4 +- .../router/SplitterMessageHandlerAdapter.java | 1 - 4 files changed, 127 insertions(+), 70 deletions(-) create mode 100644 spring-integration-adapters/src/main/java/org/springframework/integration/adapter/AbstractMessageHandlingSourceAdapter.java 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;