Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Factored out common remoting-based source adapter behavior into AbstractMessageHandlingSourceAdapter.

Browse Source
This commit is contained in:
Mark Fisher
2008-03-07 23:48:13 +00:00
parent d723fc6e24
commit 9f66c37b4c
4 changed files with 127 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/AbstractMessageHandlingSourceAdapter.java Normal file
View File

@@ -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 '<code>true</code>'.
*/
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);
}
}

70
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java
View File

@@ -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 '<code>true</code>'.
*/
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);
}
}

4
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java
View File

@@ -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);

1
spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java
View File

@@ -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;