From 0bd9ea92550f312f5dbfad813b95b2d78fdd7a34 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 17 Apr 2008 21:06:14 +0000 Subject: [PATCH] TargetEndpoint now returns false when one of its MessageSelectors rejects a Message rather than throwing an Exception. Added Source, BlockingSource, and BlockingTarget interfaces. MessageChannel now extends BlockingSource and BlockingTarget. --- .../integration/adapter/file/FileSource.java | 2 +- .../integration/adapter/ftp/FtpSource.java | 2 +- .../adapter/jms/JmsPollableSource.java | 2 +- .../adapter/stream/ByteStreamSource.java | 2 +- .../adapter/stream/CharacterStreamSource.java | 2 +- .../adapter/MethodInvokingSource.java | 2 +- .../adapter/PollingSourceAdapter.java | 2 +- .../integration/channel/MessageChannel.java | 46 +--------------- .../dispatcher/DefaultMessageDistributor.java | 11 ++-- .../dispatcher/SynchronousChannel.java | 2 +- .../integration/endpoint/TargetEndpoint.java | 3 +- .../integration/message/BlockingSource.java | 45 +++++++++++++++ .../integration/message/BlockingTarget.java | 49 +++++++++++++++++ .../integration/message/PollableSource.java | 4 +- ...ctorRejectedException.java => Source.java} | 15 +---- .../adapter/MethodInvokingSourceTests.java | 8 +-- .../adapter/PollingSourceAdapterTests.java | 2 +- .../integration/bus/MessageBusTests.java | 2 +- .../config/EndpointParserTests.java | 6 +- .../DefaultMessageDispatcherTests.java | 55 +++++++++---------- .../dispatcher/SynchronousChannelTests.java | 4 +- .../endpoint/HandlerEndpointTests.java | 37 ++++--------- 22 files changed, 163 insertions(+), 140 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/message/BlockingSource.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/message/BlockingTarget.java rename spring-integration-core/src/main/java/org/springframework/integration/message/{selector/MessageSelectorRejectedException.java => Source.java} (54%) diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java index 24eb59e3bf..c2d5a25163 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java @@ -84,7 +84,7 @@ public class FileSource implements PollableSource, InitializingBean { } } - public Message poll() { + public Message receive() { File[] files = null; if (this.fileFilter != null) { files = this.directory.listFiles(this.fileFilter); diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java index a2e1707fc6..58a6a9fc97 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java @@ -120,7 +120,7 @@ public class FtpSource implements PollableSource, MessageDeliveryAware { } } - public final Message poll() { + public final Message receive() { try { this.establishConnection(); FTPFile[] fileList = this.client.listFiles(); diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java index 2bdfdaafc2..a85db97517 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java @@ -47,7 +47,7 @@ public class JmsPollableSource extends AbstractJmsTemplateBasedAdapter implement } - public Message poll() { + public Message receive() { Object receivedObject = this.getJmsTemplate().receiveAndConvert(); if (receivedObject == null) { return null; diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java index 182ef2f638..6010667bfb 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java @@ -67,7 +67,7 @@ public class ByteStreamSource implements PollableSource { this.shouldTruncate = shouldTruncate; } - public Message poll() { + public Message receive() { try { byte[] bytes; int bytesRead = 0; diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java index 1f24502c9d..9d1c0698c3 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java @@ -58,7 +58,7 @@ public class CharacterStreamSource implements PollableSource { } - public StringMessage poll() { + public StringMessage receive() { try { synchronized (this.monitor) { if (!this.reader.ready()) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java index 3352dbabbb..5cce87c8fb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java @@ -57,7 +57,7 @@ public class MethodInvokingSource implements PollableSource, Initiali this.invoker.setMethodValidator(new MessageReceivingMethodValidator()); } - public Message poll() { + public Message receive() { if (this.invoker == null) { this.afterPropertiesSet(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java index 8d9605c7c2..d923d19c56 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -85,7 +85,7 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements Messa List> results = new ArrayList>(); int count = 0; while (count < limit) { - Message message = this.source.poll(); + Message message = this.source.receive(); if (message == null) { break; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java index 83b4f1b67e..aea26c5c2a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java @@ -18,6 +18,8 @@ package org.springframework.integration.channel; import java.util.List; +import org.springframework.integration.message.BlockingSource; +import org.springframework.integration.message.BlockingTarget; import org.springframework.integration.message.Message; import org.springframework.integration.message.selector.MessageSelector; @@ -26,7 +28,7 @@ import org.springframework.integration.message.selector.MessageSelector; * * @author Mark Fisher */ -public interface MessageChannel { +public interface MessageChannel extends BlockingSource, BlockingTarget { static final int DEFAULT_CAPACITY = 100; @@ -46,48 +48,6 @@ public interface MessageChannel { */ DispatcherPolicy getDispatcherPolicy(); - /** - * Send a message, blocking indefinitely if necessary. - * - * @param message the {@link Message} to send - * - * @return true if the message is sent successfully, - * false if interrupted - */ - boolean send(Message message); - - /** - * Send a message, blocking until either the message is accepted or the - * specified timeout period elapses. - * - * @param message the {@link Message} to send - * @param timeout the timeout in milliseconds - * - * @return true if the message is sent successfully, - * false if the specified timeout period elapses or - * the send is interrupted - */ - boolean send(Message message, long timeout); - - /** - * Receive a message, blocking indefinitely if necessary. - * - * @return the next available {@link Message} or null if - * interrupted - */ - Message receive(); - - /** - * Receive a message, blocking until either a message is available or the - * specified timeout period elapses. - * - * @param timeout the timeout in milliseconds - * - * @return the next available {@link Message} or null if the - * specified timeout period elapses or the message reception is interrupted - */ - Message receive(long timeout); - /** * Remove all {@link Message Messages} from this channel. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java index e6f36ee9c9..c29e40cbfb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java @@ -30,7 +30,6 @@ import org.springframework.integration.handler.MessageHandlerRejectedExecutionEx import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.Target; -import org.springframework.integration.message.selector.MessageSelectorRejectedException; import org.springframework.util.Assert; /** @@ -94,16 +93,14 @@ public class DefaultMessageDistributor implements MessageDistributor { if (!this.dispatcherPolicy.isPublishSubscribe() && sent) { return true; } - iter.remove(); - } - catch (MessageSelectorRejectedException e) { - if (logger.isDebugEnabled()) { - logger.debug("selector rejected message, continuing with other targets if available", e); + if (!sent && logger.isDebugEnabled()) { + logger.debug("endpoint rejected message, continuing with other targets if available"); } + iter.remove(); } catch (MessageHandlerNotRunningException e) { if (logger.isDebugEnabled()) { - logger.debug("target not running, continuing with other targets if available", e); + logger.debug("target is not running, continuing with other targets if available", e); } } catch (MessageHandlerRejectedExecutionException e) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java index 883e434b52..c2eda45b01 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java @@ -88,7 +88,7 @@ public class SynchronousChannel extends AbstractMessageChannel { @Override protected Message doReceive(long timeout) { if (this.source != null) { - Message result = this.source.poll(); + Message result = this.source.receive(); if (result != null) { return result; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java index 2591f47f0f..6396579c53 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java @@ -38,7 +38,6 @@ import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.Target; import org.springframework.integration.message.selector.MessageSelector; -import org.springframework.integration.message.selector.MessageSelectorRejectedException; import org.springframework.integration.scheduling.Subscription; import org.springframework.integration.util.ErrorHandler; import org.springframework.util.Assert; @@ -205,7 +204,7 @@ public class TargetEndpoint implements MessageEndpoint, BeanNameAware { } for (MessageSelector selector : this.selectors) { if (!selector.accept(message)) { - throw new MessageSelectorRejectedException(message); + return false; } } try { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/BlockingSource.java b/spring-integration-core/src/main/java/org/springframework/integration/message/BlockingSource.java new file mode 100644 index 0000000000..83e726e702 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/BlockingSource.java @@ -0,0 +1,45 @@ +/* + * 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; + +/** + * Extends {@link PollableSource} and provides a timeout-aware receive method. + * + * @author Mark Fisher + */ +public interface BlockingSource extends PollableSource { + + /** + * Receive a message, blocking indefinitely if necessary. + * + * @return the next available {@link Message} or null if + * interrupted + */ + Message receive(); + + /** + * Receive a message, blocking until either a message is available or the + * specified timeout period elapses. + * + * @param timeout the timeout in milliseconds + * + * @return the next available {@link Message} or null if the + * specified timeout period elapses or the message reception is interrupted + */ + Message receive(long timeout); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/BlockingTarget.java b/spring-integration-core/src/main/java/org/springframework/integration/message/BlockingTarget.java new file mode 100644 index 0000000000..6ecffacdb9 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/BlockingTarget.java @@ -0,0 +1,49 @@ +/* + * 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; + +/** + * Extends {@link Target} and provides a timeout-aware send method. + * + * @author Mark Fisher + */ +public interface BlockingTarget extends Target { + + /** + * Send a message, blocking indefinitely if necessary. + * + * @param message the {@link Message} to send + * + * @return true if the message is sent successfully, + * false if interrupted + */ + boolean send(Message message); + + /** + * Send a message, blocking until either the message is accepted or the + * specified timeout period elapses. + * + * @param message the {@link Message} to send + * @param timeout the timeout in milliseconds + * + * @return true if the message is sent successfully, + * false if the specified timeout period elapses or + * the send is interrupted + */ + boolean send(Message message, long timeout); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java b/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java index eb50874665..f8d39416d1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java @@ -21,11 +21,11 @@ package org.springframework.integration.message; * * @author Mark Fisher */ -public interface PollableSource { +public interface PollableSource extends Source { /** * Retrieve a message from this source or null if no message is available. */ - Message poll(); + Message receive(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/selector/MessageSelectorRejectedException.java b/spring-integration-core/src/main/java/org/springframework/integration/message/Source.java similarity index 54% rename from spring-integration-core/src/main/java/org/springframework/integration/message/selector/MessageSelectorRejectedException.java rename to spring-integration-core/src/main/java/org/springframework/integration/message/Source.java index 9a8aaa63e0..4357a8c6d4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/selector/MessageSelectorRejectedException.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/Source.java @@ -14,22 +14,13 @@ * limitations under the License. */ -package org.springframework.integration.message.selector; - -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageHandlingException; +package org.springframework.integration.message; /** - * An exception indicating that a message was rejected by an implementation of - * {@link org.springframework.integration.message.selector.MessageSelector}. + * Base interface for any source of {@link Message Messages}. * * @author Mark Fisher */ -@SuppressWarnings("serial") -public class MessageSelectorRejectedException extends MessageHandlingException { - - public MessageSelectorRejectedException(Message rejectedMessage) { - super(rejectedMessage); - } +public interface Source { } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java index 8f6a826af3..ae0d99ec61 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java @@ -34,7 +34,7 @@ public class MethodInvokingSourceTests { MethodInvokingSource source = new MethodInvokingSource(); source.setObject(new TestBean()); source.setMethod("validMethod"); - Message result = source.poll(); + Message result = source.receive(); assertNotNull(result); assertNotNull(result.getPayload()); assertEquals("valid", result.getPayload()); @@ -45,7 +45,7 @@ public class MethodInvokingSourceTests { MethodInvokingSource source = new MethodInvokingSource(); source.setObject(new TestBean()); source.setMethod("noMatchingMethod"); - source.poll(); + source.receive(); } @Test(expected=MessagingException.class) @@ -53,7 +53,7 @@ public class MethodInvokingSourceTests { MethodInvokingSource source = new MethodInvokingSource(); source.setObject(new TestBean()); source.setMethod("invalidMethodWithArg"); - source.poll(); + source.receive(); } @Test(expected=MessagingException.class) @@ -61,7 +61,7 @@ public class MethodInvokingSourceTests { MethodInvokingSource source = new MethodInvokingSource(); source.setObject(new TestBean()); source.setMethod("invalidMethodWithNoReturnValue"); - source.poll(); + source.receive(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java index 04f1553cea..470fb416bf 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java @@ -108,7 +108,7 @@ public class PollingSourceAdapterTests { this.count.set(0); } - public Message poll() { + public Message receive() { if (count.get() >= limit) { return null; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index f240a5bcb4..809ce44c9b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -268,7 +268,7 @@ public class MessageBusTests { this.latch = latch; } - public Message poll() { + public Message receive() { latch.countDown(); throw new RuntimeException("intentional test failure"); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java index 18bdbfe477..cc4a0dc2c0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; @@ -35,7 +36,6 @@ import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.Target; -import org.springframework.integration.message.selector.MessageSelectorRejectedException; /** * @author Mark Fisher @@ -132,13 +132,13 @@ public class EndpointParserTests { assertEquals("foo", reply.getPayload()); } - @Test(expected=MessageSelectorRejectedException.class) + @Test public void testEndpointWithSelectorRejects() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "endpointWithSelectors.xml", this.getClass()); Target endpoint = (Target) context.getBean("endpoint"); ((Lifecycle) endpoint).start(); - endpoint.send(new GenericMessage(123)); + assertFalse(endpoint.send(new GenericMessage(123))); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java index 8a67d4fabc..7cad05f5e6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java @@ -384,48 +384,43 @@ public class DefaultMessageDispatcherTests { public void testTwoExecutorsWithSelectorsAndNeitherAccepts() throws InterruptedException { final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); - final AtomicInteger attemptedCounter1 = new AtomicInteger(); - final AtomicInteger attemptedCounter2 = new AtomicInteger(); - final CountDownLatch attemptedLatch = new CountDownLatch(2); + final AtomicInteger selectorCounter1 = new AtomicInteger(); + final AtomicInteger selectorCounter2 = new AtomicInteger(); + final CountDownLatch selectorLatch = new CountDownLatch(2); final CountDownLatch handlerLatch = new CountDownLatch(1); - MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, attemptedLatch); - MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, attemptedLatch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, handlerLatch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, handlerLatch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); final HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); final HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2); - endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class)); - endpoint2.addMessageSelector(new PayloadTypeSelector(Integer.class)); - endpoint1.start(); - endpoint2.start(); - MessageHandler interceptor1 = new MessageHandler() { - public Message handle(Message message) { - attemptedCounter1.incrementAndGet(); - attemptedLatch.countDown(); - endpoint1.send(message); - return null; + endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class) { + @Override + public boolean accept(Message message) { + selectorCounter1.incrementAndGet(); + selectorLatch.countDown(); + return super.accept(message); } - }; - MessageHandler interceptor2 = new MessageHandler() { - public Message handle(Message message) { - attemptedCounter2.incrementAndGet(); - attemptedLatch.countDown(); - endpoint2.send(message); - return null; + }); + endpoint2.addMessageSelector(new PayloadTypeSelector(Integer.class) { + @Override + public boolean accept(Message message) { + selectorCounter2.incrementAndGet(); + selectorLatch.countDown(); + return super.accept(message); } - }; - dispatcher.addTarget(createEndpoint(interceptor1, false)); - dispatcher.addTarget(createEndpoint(interceptor2, false)); + }); + dispatcher.addTarget(endpoint1); + dispatcher.addTarget(endpoint2); dispatcher.start(); - attemptedLatch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, attemptedLatch.getCount()); + selectorLatch.await(2000, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, selectorLatch.getCount()); assertEquals("handler1 should not have accepted the message", 0, counter1.get()); assertEquals("handler2 should not have accepted the message", 0, counter2.get()); - assertEquals("executor1 should have had exactly one attempt", 1, attemptedCounter1.get()); - assertEquals("executor2 should have had exactly one attempt", 1, attemptedCounter2.get()); + assertEquals("executor1 should have had exactly one attempt", 1, selectorCounter1.get()); + assertEquals("executor2 should have had exactly one attempt", 1, selectorCounter2.get()); assertEquals("handlerLatch should not have counted down", 1, handlerLatch.getCount()); - assertEquals("attemptedLatch should have counted down", 0, attemptedLatch.getCount()); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java index 989e962330..26b7b87796 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java @@ -97,7 +97,7 @@ public class SynchronousChannelTests { @Test public void testReceive() { SynchronousChannel channel = new SynchronousChannel(new PollableSource() { - public Message poll() { + public Message receive() { return new StringMessage("foo"); } }); @@ -184,7 +184,7 @@ public class SynchronousChannelTests { this.messageText = messageText; } - public StringMessage poll() { + public StringMessage receive() { StringMessage message = new StringMessage(messageText); message.getHeader().setProperty(HANDLER_THREAD, Thread.currentThread().getName()); return message; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java index 58df061cf7..3a4c97b6a8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java @@ -39,7 +39,6 @@ import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.selector.MessageSelector; -import org.springframework.integration.message.selector.MessageSelectorRejectedException; import org.springframework.integration.util.ErrorHandler; /** @@ -336,7 +335,7 @@ public class HandlerEndpointTests { assertTrue(exceptionThrown); } - @Test(expected=MessageSelectorRejectedException.class) + @Test public void testEndpointWithSelectorRejecting() { HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.nullHandler()); endpoint.addMessageSelector(new MessageSelector() { @@ -345,7 +344,7 @@ public class HandlerEndpointTests { } }); endpoint.start(); - endpoint.send(new StringMessage("test")); + assertFalse(endpoint.send(new StringMessage("test"))); } @Test @@ -368,7 +367,6 @@ public class HandlerEndpointTests { public void testEndpointWithMultipleSelectorsAndFirstRejects() { final AtomicInteger counter = new AtomicInteger(); HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(counter)); - boolean exceptionThrown = false; endpoint.addMessageSelector(new MessageSelector() { public boolean accept(Message message) { counter.incrementAndGet(); @@ -382,43 +380,32 @@ public class HandlerEndpointTests { } }); endpoint.start(); - try { - endpoint.send(new StringMessage("test")); - } - catch (MessageSelectorRejectedException e) { - exceptionThrown = true; - } + assertFalse(endpoint.send(new StringMessage("test"))); assertEquals("only the first selector should have been invoked", 1, counter.get()); - assertTrue(exceptionThrown); endpoint.stop(); } @Test public void testEndpointWithMultipleSelectorsAndFirstAccepts() { - final AtomicInteger counter = new AtomicInteger(); - HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(counter)); - boolean exceptionThrown = false; + final AtomicInteger selectorCounter = new AtomicInteger(); + AtomicInteger handlerCounter = new AtomicInteger(); + HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(handlerCounter)); endpoint.addMessageSelector(new MessageSelector() { public boolean accept(Message message) { - counter.incrementAndGet(); + selectorCounter.incrementAndGet(); return true; } }); endpoint.addMessageSelector(new MessageSelector() { public boolean accept(Message message) { - counter.incrementAndGet(); + selectorCounter.incrementAndGet(); return false; } }); endpoint.start(); - try { - endpoint.send(new StringMessage("test")); - } - catch (MessageSelectorRejectedException e) { - exceptionThrown = true; - } - assertEquals("both selectors should have been invoked but not the handler", 2, counter.get()); - assertTrue(exceptionThrown); + assertFalse(endpoint.send(new StringMessage("test"))); + assertEquals("both selectors should have been invoked", 2, selectorCounter.get()); + assertEquals("the handler should not have been invoked", 0, handlerCounter.get()); endpoint.stop(); } @@ -439,7 +426,7 @@ public class HandlerEndpointTests { } }); endpoint.start(); - endpoint.send(new StringMessage("test")); + assertTrue(endpoint.send(new StringMessage("test"))); assertEquals("both selectors and handler should have been invoked", 3, counter.get()); endpoint.stop(); }