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

Refactoring lifecycle code for endpoint hierarchy.

Browse Source
This commit is contained in:
Mark Fisher
2008-07-01 23:01:42 +00:00
parent 9dad71117b
commit 570cee8f9b
4 changed files with 81 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
View File

@@ -25,7 +25,10 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.handler.MessageHandlerNotRunningException;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
@@ -34,7 +37,7 @@ import org.springframework.integration.message.MessageHandlingException;
*
* @author Mark Fisher
*/
public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware {
public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware, Lifecycle, InitializingBean {
protected final Log logger = LogFactory.getLog(this.getClass());
@@ -42,6 +45,12 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
private final List<Advice> interceptors = new ArrayList<Advice>();
private volatile boolean autoStartup = true;
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
public String getName() {
return this.name;
@@ -55,8 +64,8 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
this.setName(beanName);
}
public String toString() {
return (this.name != null) ? this.name : super.toString();
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
public void addInterceptor(Object interceptor) {
@@ -83,10 +92,55 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
return this.interceptors;
}
public String toString() {
return (this.name != null) ? this.name : super.toString();
}
protected void initialize() {
}
public boolean isRunning() {
return this.running;
}
public void afterPropertiesSet() {
if (this.autoStartup) {
this.start();
}
else {
this.initialize();
}
}
public void start() {
this.initialize();
synchronized (this.lifecycleMonitor) {
if (this.running) {
return;
}
this.running = true;
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (!this.running) {
return;
}
this.running = false;
}
}
public final boolean invoke(Message<?> message) {
if (message == null) {
throw new IllegalArgumentException("Message must not be null.");
}
if (logger.isDebugEnabled()) {
logger.debug("endpoint '" + this + "' handling message: " + message);
}
if (!this.isRunning()) {
throw new MessageHandlerNotRunningException(message);
}
if (!this.supports(message)) {
throw new MessageHandlingException(message, "unsupported message");
}

4
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java
View File

@@ -85,13 +85,13 @@ public class HandlerEndpoint extends TargetEndpoint implements MessageProducingE
this.returnAddressOverrides = returnAddressOverrides;
}
public void afterPropertiesSet() {
public void initialize() {
Assert.notNull(this.handler, "handler must not be null");
if (this.handler instanceof ChannelRegistryAware) {
((ChannelRegistryAware) this.handler).setChannelRegistry(this.getChannelRegistry());
}
super.setTarget(new HandlerInvokingTarget(this.handler, this.replyHandler));
super.afterPropertiesSet();
super.initialize();
}
private MessageChannel resolveReplyChannel(MessageHeader originalMessageHeader) {

58
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java
View File

@@ -16,12 +16,8 @@
package org.springframework.integration.endpoint;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.handler.MessageHandlerNotRunningException;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.selector.MessageSelector;
@@ -34,7 +30,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
*/
public class TargetEndpoint extends AbstractEndpoint
implements MessageTarget, MessageConsumingEndpoint, ChannelRegistryAware, InitializingBean, Lifecycle {
implements MessageTarget, MessageConsumingEndpoint, ChannelRegistryAware {
private volatile MessageTarget target;
@@ -46,7 +42,7 @@ public class TargetEndpoint extends AbstractEndpoint
private volatile boolean initialized;
private volatile boolean running;
private final Object initializationMonitor = new Object();
public TargetEndpoint() {
@@ -90,52 +86,20 @@ public class TargetEndpoint extends AbstractEndpoint
return this.channelRegistry;
}
public void afterPropertiesSet() {
if (this.target instanceof ChannelRegistryAware && this.channelRegistry != null) {
((ChannelRegistryAware) this.target).setChannelRegistry(this.channelRegistry);
protected void initialize() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
}
if (this.target instanceof ChannelRegistryAware && this.channelRegistry != null) {
((ChannelRegistryAware) this.target).setChannelRegistry(this.channelRegistry);
}
this.initialized = true;
}
this.initialized = true;
}
public boolean isRunning() {
return this.running;
}
public void start() {
if (this.isRunning()) {
return;
}
if (!this.initialized) {
this.afterPropertiesSet();
}
this.running = true;
}
public void stop() {
if (!this.isRunning()) {
return;
}
if (this.target instanceof DisposableBean) {
try {
((DisposableBean) this.target).destroy();
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("exception occurred when destroying target", e);
}
}
}
this.running = false;
}
@Override
protected final boolean doInvoke(Message<?> message) {
if (logger.isDebugEnabled()) {
logger.debug("endpoint '" + this + "' handling message: " + message);
}
if (!this.isRunning()) {
throw new MessageHandlerNotRunningException(message);
}
if (this.selector != null && !this.selector.accept(message)) {
return false;
}

11
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java
View File

@@ -27,6 +27,7 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.message.CommandMessage;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.PollCommand;
import org.springframework.integration.message.MessageSource;
@@ -40,12 +41,22 @@ public class SourceEndpointTests {
TestSource source = new TestSource("testing", 1);
QueueChannel channel = new QueueChannel();
SourceEndpoint endpoint = new SourceEndpoint(source, channel);
endpoint.afterPropertiesSet();
endpoint.invoke(new CommandMessage(new PollCommand()));
Message<?> message = channel.receive(1000);
assertNotNull("message should not be null", message);
assertEquals("testing.1", message.getPayload());
}
@Test(expected = MessageHandlingException.class)
public void testAutoStartupDisabled() {
TestSource source = new TestSource("testing", 1);
QueueChannel channel = new QueueChannel();
SourceEndpoint endpoint = new SourceEndpoint(source, channel);
endpoint.setAutoStartup(false);
endpoint.afterPropertiesSet();
endpoint.invoke(new CommandMessage(new PollCommand()));
}
private static class TestSource implements MessageSource<String> {