diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java index 68222ec41c..9785eade90 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java @@ -25,10 +25,10 @@ import org.junit.Test; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.EndpointPoller; import org.springframework.integration.endpoint.SourceEndpoint; -import org.springframework.integration.message.CommandMessage; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.PollCommand; import org.springframework.integration.scheduling.PollingSchedule; /** @@ -45,7 +45,7 @@ public class ByteStreamSourceTests { SourceEndpoint endpoint = new SourceEndpoint(source); endpoint.setOutputChannel(channel); endpoint.afterPropertiesSet(); - endpoint.send(new CommandMessage(new PollCommand())); + endpoint.send(new GenericMessage(new EndpointPoller())); Message message1 = channel.receive(500); byte[] payload = (byte[]) message1.getPayload(); assertEquals(3, payload.length); @@ -54,7 +54,7 @@ public class ByteStreamSourceTests { assertEquals(3, payload[2]); Message message2 = channel.receive(0); assertNull(message2); - endpoint.send(new CommandMessage(new PollCommand())); + endpoint.send(new GenericMessage(new EndpointPoller())); Message message3 = channel.receive(0); assertNull(message3); } @@ -71,12 +71,12 @@ public class ByteStreamSourceTests { SourceEndpoint endpoint = new SourceEndpoint(source); endpoint.setOutputChannel(channel); endpoint.afterPropertiesSet(); - endpoint.send(new CommandMessage(new PollCommand())); + endpoint.send(new GenericMessage(new EndpointPoller())); Message message1 = channel.receive(0); assertEquals(4, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); assertNull(message2); - endpoint.send(new CommandMessage(new PollCommand())); + endpoint.send(new GenericMessage(new EndpointPoller())); Message message3 = channel.receive(0); assertEquals(2, ((byte[]) message3.getPayload()).length); } @@ -94,12 +94,12 @@ public class ByteStreamSourceTests { SourceEndpoint endpoint = new SourceEndpoint(source); endpoint.setOutputChannel(channel); endpoint.afterPropertiesSet(); - endpoint.send(new CommandMessage(new PollCommand())); + endpoint.send(new GenericMessage(new EndpointPoller())); Message message1 = channel.receive(0); assertEquals(4, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); assertNull(message2); - endpoint.send(new CommandMessage(new PollCommand())); + endpoint.send(new GenericMessage(new EndpointPoller())); Message message3 = channel.receive(0); assertEquals(4, ((byte[]) message3.getPayload()).length); assertEquals(0, ((byte[]) message3.getPayload())[3]); diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java index 007b24ae91..0ebd999229 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java @@ -25,10 +25,10 @@ import org.junit.Test; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.EndpointPoller; import org.springframework.integration.endpoint.SourceEndpoint; -import org.springframework.integration.message.CommandMessage; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.PollCommand; import org.springframework.integration.scheduling.PollingSchedule; /** @@ -46,12 +46,12 @@ public class CharacterStreamSourceTests { SourceEndpoint endpoint = new SourceEndpoint(source); endpoint.setOutputChannel(channel); endpoint.afterPropertiesSet(); - endpoint.send(new CommandMessage(new PollCommand())); + endpoint.send(new GenericMessage(new EndpointPoller())); Message message1 = channel.receive(0); assertEquals("test", message1.getPayload()); Message message2 = channel.receive(0); assertNull(message2); - endpoint.send(new CommandMessage(new PollCommand())); + endpoint.send(new GenericMessage(new EndpointPoller())); Message message3 = channel.receive(0); assertNull(message3); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index 3a30a430e0..bc24696bbc 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -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); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/CommandMessage.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointPoller.java similarity index 78% rename from org.springframework.integration/src/main/java/org/springframework/integration/message/CommandMessage.java rename to org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointPoller.java index a09a0b20ce..c06c80d1e4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/CommandMessage.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointPoller.java @@ -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 { +public class EndpointPoller implements EndpointVisitor { - public CommandMessage(Command command) { - super(command); + public void visitEndpoint(MessageEndpoint endpoint) { + endpoint.poll(); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java index a483fcf61d..a68d95d22e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java @@ -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 { + private static class EndpointPollerMessageSource implements MessageSource { - public Message receive() { - return new GenericMessage(new PollCommand()); + public Message receive() { + return new GenericMessage(new EndpointPoller()); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/Command.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointVisitor.java similarity index 83% rename from org.springframework.integration/src/main/java/org/springframework/integration/message/Command.java rename to org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointVisitor.java index ec9cbc3b3c..259a59c202 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/Command.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointVisitor.java @@ -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); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index d12d0656fb..810cc2c1ce 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -45,4 +45,6 @@ public interface MessageEndpoint extends MessageTarget, ChannelRegistryAware, In MessageChannel getOutputChannel(); + boolean poll(); + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java index 1bf320963b..8a9fdb73ea 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java @@ -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; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java index f27205a9f1..4b52d20db1 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java @@ -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; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/PollCommand.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/PollCommand.java deleted file mode 100644 index edd2938a57..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/PollCommand.java +++ /dev/null @@ -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 { - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageEndpointBeanPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageEndpointBeanPostProcessorTests.java index 7d5644a200..b1312c3859 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageEndpointBeanPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageEndpointBeanPostProcessorTests.java @@ -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(new EndpointPoller())); assertEquals(1, beforeAdvice.getCount()); assertEquals(2, interceptor.getCount()); context.stop(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java index 41a4cb5ba1..57794ddb9d 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java @@ -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(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(new EndpointPoller())); } private static class TestSource implements MessageSource {