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

Endpoints now recognize an EndpointVisitor in the message payload. The EndpointPoller is now an implementation of EndpointVisitor (removed PollCommand).

Browse Source
This commit is contained in:
Mark Fisher
2008-07-06 03:09:56 +00:00
parent 06f4e1d011
commit 1b90086e4e
12 changed files with 49 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -30,7 +30,6 @@ import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.MessageHandlerNotRunningException;
import org.springframework.integration.message.Command;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.scheduling.Schedule;
@@ -231,8 +230,9 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
if (!this.isRunning()) {
throw new MessageHandlerNotRunningException(message);
}
if (message.getPayload() instanceof Command) {
return this.handleCommand((Command) message.getPayload());
if (message.getPayload() instanceof EndpointVisitor) {
((EndpointVisitor) message.getPayload()).visitEndpoint(this);
return true;
}
if (!this.supports(message)) {
throw new MessageRejectedException(message, "unsupported message");
@@ -242,8 +242,6 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
protected abstract boolean supports(Message<?> message);
protected abstract boolean handleCommand(Command command);
protected abstract boolean handleMessage(Message<?> message);
}

8
org.springframework.integration/src/main/java/org/springframework/integration/message/CommandMessage.java → org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointPoller.java
View File

@@ -14,15 +14,15 @@
* limitations under the License.
*/
package org.springframework.integration.message;
package org.springframework.integration.endpoint;
/**
* @author Mark Fisher
*/
public class CommandMessage extends GenericMessage<Command> {
public class EndpointPoller implements EndpointVisitor {
public CommandMessage(Command command) {
super(command);
public void visitEndpoint(MessageEndpoint endpoint) {
endpoint.poll();
}
}

19
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java
View File

@@ -21,38 +21,37 @@ import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.PollCommand;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
/**
* A {@link PollingDispatcher} implementation that sends poll command
* trigger messages to endpoints.
* A {@link PollingDispatcher} implementation that sends a message
* to trigger endpoint polling.
*
* @author Mark Fisher
*/
public class EndpointTrigger extends PollingDispatcher {
/**
* Create a trigger with the specified {@link Schedule}.
* Create an endpoint trigger with the specified {@link Schedule}.
*/
public EndpointTrigger(Schedule schedule) {
super(new PollCommandMessageSource(), new BroadcastingDispatcher(), schedule);
super(new EndpointPollerMessageSource(), new BroadcastingDispatcher(), schedule);
}
/**
* Create a trigger. A {@link PollingSchedule} will be created
* with the specified interval.
* Create an endpoint trigger. A {@link PollingSchedule} will be
* created with the specified interval.
*/
public EndpointTrigger(long interval) {
this(new PollingSchedule(interval));
}
private static class PollCommandMessageSource implements MessageSource<PollCommand> {
private static class EndpointPollerMessageSource implements MessageSource<EndpointPoller> {
public Message<PollCommand> receive() {
return new GenericMessage<PollCommand>(new PollCommand());
public Message<EndpointPoller> receive() {
return new GenericMessage<EndpointPoller>(new EndpointPoller());
}
}

8
org.springframework.integration/src/main/java/org/springframework/integration/message/Command.java → org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointVisitor.java
View File

@@ -14,13 +14,13 @@
* limitations under the License.
*/
package org.springframework.integration.message;
package org.springframework.integration.endpoint;
/**
* A marker interface for commands.
*
* @author Mark Fisher
*/
public interface Command {
public interface EndpointVisitor {
void visitEndpoint(MessageEndpoint endpoint);
}

2
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
View File

@@ -45,4 +45,6 @@ public interface MessageEndpoint extends MessageTarget, ChannelRegistryAware, In
MessageChannel getOutputChannel();
boolean poll();
}

12
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java
View File

@@ -18,12 +18,10 @@ package org.springframework.integration.endpoint;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Command;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryAware;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.PollCommand;
import org.springframework.util.Assert;
/**
@@ -61,15 +59,7 @@ public class SourceEndpoint extends AbstractEndpoint {
return false;
}
@Override
protected final boolean handleCommand(Command command) {
if (command instanceof PollCommand) {
return this.poll();
}
return false;
}
private boolean poll() {
public boolean poll() {
Message<?> message = this.source.receive();
if (message == null) {
return false;

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

@@ -19,10 +19,8 @@ package org.springframework.integration.endpoint;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Command;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.PollCommand;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.util.Assert;
@@ -105,20 +103,17 @@ public class TargetEndpoint extends AbstractEndpoint {
((BlockingTarget) this.target).send(message) : this.target.send(message);
}
@Override
protected final boolean handleCommand(Command command) {
if (command instanceof PollCommand) {
MessageChannel channel = this.getInputChannel();
if (channel != null) {
Message<?> receivedMessage = channel.receive(this.receiveTimeout);
if (receivedMessage != null) {
return this.handleMessage(receivedMessage);
}
}
else if (logger.isDebugEnabled()) {
logger.debug("TargetEndpoint unable to resolve channel '" + this.getInputChannelName() + "'");
public final boolean poll() {
MessageChannel channel = this.getInputChannel();
if (channel != null) {
Message<?> receivedMessage = channel.receive(this.receiveTimeout);
if (receivedMessage != null) {
return this.handleMessage(receivedMessage);
}
}
else if (logger.isDebugEnabled()) {
logger.debug("TargetEndpoint unable to resolve channel '" + this.getInputChannelName() + "'");
}
return false;
}

24
org.springframework.integration/src/main/java/org/springframework/integration/message/PollCommand.java
View File

@@ -1,24 +0,0 @@
/*
* 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.message;
/**
* @author Mark Fisher
*/
public class PollCommand implements Command {
}

6
org.springframework.integration/src/test/java/org/springframework/integration/config/MessageEndpointBeanPostProcessorTests.java
View File

@@ -25,9 +25,9 @@ import org.junit.Test;
import org.springframework.aop.support.AopUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.endpoint.EndpointPoller;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.message.CommandMessage;
import org.springframework.integration.message.PollCommand;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.StringMessage;
/**
@@ -101,7 +101,7 @@ public class MessageEndpointBeanPostProcessorTests {
TestEndpointInterceptor interceptor = (TestEndpointInterceptor) context.getBean("interceptor");
assertEquals(0, beforeAdvice.getCount());
assertEquals(0, interceptor.getCount());
endpoint.send(new CommandMessage(new PollCommand()));
endpoint.send(new GenericMessage<EndpointPoller>(new EndpointPoller()));
assertEquals(1, beforeAdvice.getCount());
assertEquals(2, interceptor.getCount());
context.stop();

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

@@ -24,11 +24,9 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
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;
/**
@@ -43,7 +41,7 @@ public class SourceEndpointTests {
SourceEndpoint endpoint = new SourceEndpoint(source);
endpoint.setOutputChannel(channel);
endpoint.afterPropertiesSet();
endpoint.send(new CommandMessage(new PollCommand()));
endpoint.send(new GenericMessage<EndpointPoller>(new EndpointPoller()));
Message<?> message = channel.receive(1000);
assertNotNull("message should not be null", message);
assertEquals("testing.1", message.getPayload());
@@ -57,7 +55,7 @@ public class SourceEndpointTests {
endpoint.setOutputChannel(channel);
endpoint.setAutoStartup(false);
endpoint.afterPropertiesSet();
endpoint.send(new CommandMessage(new PollCommand()));
endpoint.send(new GenericMessage<EndpointPoller>(new EndpointPoller()));
}
private static class TestSource implements MessageSource<String> {