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

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.

Browse Source
This commit is contained in:
Mark Fisher
2008-04-17 21:06:14 +00:00
parent 0f6bd60ee0
commit 0bd9ea9255
22 changed files with 163 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java
View File

@@ -84,7 +84,7 @@ public class FileSource implements PollableSource<Object>, InitializingBean {
}
}
public Message poll() {
public Message receive() {
File[] files = null;
if (this.fileFilter != null) {
files = this.directory.listFiles(this.fileFilter);

2
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java
View File

@@ -120,7 +120,7 @@ public class FtpSource implements PollableSource<Object>, MessageDeliveryAware {
}
}
public final Message poll() {
public final Message receive() {
try {
this.establishConnection();
FTPFile[] fileList = this.client.listFiles();

2
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java
View File

@@ -47,7 +47,7 @@ public class JmsPollableSource extends AbstractJmsTemplateBasedAdapter implement
}
public Message<Object> poll() {
public Message<Object> receive() {
Object receivedObject = this.getJmsTemplate().receiveAndConvert();
if (receivedObject == null) {
return null;

2
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java
View File

@@ -67,7 +67,7 @@ public class ByteStreamSource implements PollableSource<byte[]> {
this.shouldTruncate = shouldTruncate;
}
public Message<byte[]> poll() {
public Message<byte[]> receive() {
try {
byte[] bytes;
int bytesRead = 0;

2
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java
View File

@@ -58,7 +58,7 @@ public class CharacterStreamSource implements PollableSource<String> {
}
public StringMessage poll() {
public StringMessage receive() {
try {
synchronized (this.monitor) {
if (!this.reader.ready()) {

2
spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java
View File

@@ -57,7 +57,7 @@ public class MethodInvokingSource<T> implements PollableSource<Object>, Initiali
this.invoker.setMethodValidator(new MessageReceivingMethodValidator());
}
public Message<Object> poll() {
public Message<Object> receive() {
if (this.invoker == null) {
this.afterPropertiesSet();
}

2
spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java
View File

@@ -85,7 +85,7 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements Messa
List<Message<?>> results = new ArrayList<Message<?>>();
int count = 0;
while (count < limit) {
Message<?> message = this.source.poll();
Message<?> message = this.source.receive();
if (message == null) {
break;
}

46
spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java
View File

@@ -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 <code>true</code> if the message is sent successfully,
* <code>false</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 <code>true</code> if the message is sent successfully,
* <code>false</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 <code>null</code> 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 <code>null</code> if the
* specified timeout period elapses or the message reception is interrupted
*/
Message<?> receive(long timeout);
/**
* Remove all {@link Message Messages} from this channel.
*/

11
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java
View File

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

2
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java
View File

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

3
spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java
View File

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

45
spring-integration-core/src/main/java/org/springframework/integration/message/BlockingSource.java Normal file
View File

@@ -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<T> extends PollableSource<T> {
/**
* Receive a message, blocking indefinitely if necessary.
*
* @return the next available {@link Message} or <code>null</code> if
* interrupted
*/
Message<T> 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 <code>null</code> if the
* specified timeout period elapses or the message reception is interrupted
*/
Message<T> receive(long timeout);
}

49
spring-integration-core/src/main/java/org/springframework/integration/message/BlockingTarget.java Normal file
View File

@@ -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 <code>true</code> if the message is sent successfully,
* <code>false</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 <code>true</code> if the message is sent successfully,
* <code>false</false> if the specified timeout period elapses or
* the send is interrupted
*/
boolean send(Message<?> message, long timeout);
}

4
spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java
View File

@@ -21,11 +21,11 @@ package org.springframework.integration.message;
*
* @author Mark Fisher
*/
public interface PollableSource<T> {
public interface PollableSource<T> extends Source {
/**
* Retrieve a message from this source or <code>null</code> if no message is available.
*/
Message<T> poll();
Message<T> receive();
}

15
spring-integration-core/src/main/java/org/springframework/integration/message/selector/MessageSelectorRejectedException.java → spring-integration-core/src/main/java/org/springframework/integration/message/Source.java
View File

@@ -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 {
}

8
spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java
View File

@@ -34,7 +34,7 @@ public class MethodInvokingSourceTests {
MethodInvokingSource<TestBean> source = new MethodInvokingSource<TestBean>();
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<TestBean> source = new MethodInvokingSource<TestBean>();
source.setObject(new TestBean());
source.setMethod("noMatchingMethod");
source.poll();
source.receive();
}
@Test(expected=MessagingException.class)
@@ -53,7 +53,7 @@ public class MethodInvokingSourceTests {
MethodInvokingSource<TestBean> source = new MethodInvokingSource<TestBean>();
source.setObject(new TestBean());
source.setMethod("invalidMethodWithArg");
source.poll();
source.receive();
}
@Test(expected=MessagingException.class)
@@ -61,7 +61,7 @@ public class MethodInvokingSourceTests {
MethodInvokingSource<TestBean> source = new MethodInvokingSource<TestBean>();
source.setObject(new TestBean());
source.setMethod("invalidMethodWithNoReturnValue");
source.poll();
source.receive();
}

2
spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java
View File

@@ -108,7 +108,7 @@ public class PollingSourceAdapterTests {
this.count.set(0);
}
public Message<String> poll() {
public Message<String> receive() {
if (count.get() >= limit) {
return null;
}

2
spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
View File

@@ -268,7 +268,7 @@ public class MessageBusTests {
this.latch = latch;
}
public Message<Object> poll() {
public Message<Object> receive() {
latch.countDown();
throw new RuntimeException("intentional test failure");
}

6
spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java
View File

@@ -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<Integer>(123));
assertFalse(endpoint.send(new GenericMessage<Integer>(123)));
}
@Test

55
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java
View File

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

4
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java
View File

@@ -97,7 +97,7 @@ public class SynchronousChannelTests {
@Test
public void testReceive() {
SynchronousChannel channel = new SynchronousChannel(new PollableSource<String>() {
public Message<String> poll() {
public Message<String> 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;

37
spring-integration-core/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java
View File

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