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

INT-886 Failed dispatching due to "no subscribers" now provides a more meaningful exception message.

Browse Source
This commit is contained in:
Mark Fisher
2009-12-10 00:13:48 +00:00
parent 24201290e6
commit 538b13b757
7 changed files with 50 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -25,7 +25,9 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.core.MessageHistory.ComponentType;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.util.Assert;
/**
@@ -117,9 +119,18 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName
if (message == null) {
return false;
}
boolean sent = this.doSend(message, timeout);
this.interceptors.postSend(message, this, sent);
return sent;
try {
boolean sent = this.doSend(message, timeout);
this.interceptors.postSend(message, this, sent);
return sent;
}
catch (MessagingException e) {
throw e;
}
catch (Exception e) {
throw new MessageDeliveryException(message,
"failed to send Message to channel '" + this.getName() + "'", e);
}
}
public String toString() {

2
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java
View File

@@ -94,7 +94,7 @@ public class UnicastingDispatcher extends AbstractDispatcher {
boolean success = false;
Iterator<MessageHandler> handlerIterator = this.getHandlerIterator(message);
if (!handlerIterator.hasNext()) {
throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
throw new IllegalStateException("Dispatcher has no subscribers.");
}
List<RuntimeException> exceptions = new ArrayList<RuntimeException>();
while (success == false && handlerIterator.hasNext()) {

4
org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java
View File

@@ -38,8 +38,8 @@ import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.util.Assert;
/**
@@ -206,7 +206,7 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware,
releaseMessage(message);
}
catch (Exception e) {
Exception exception = new MessageDeliveryException(message, "Failed to deliver Message after delay.", e);
Exception exception = new MessageHandlingException(message, "Failed to deliver Message after delay.", e);
MessageChannel errorChannel = resolveErrorChannelIfPossible(message);
if (errorChannel != null) {
ErrorMessage errorMessage = new ErrorMessage(exception);

3
org.springframework.integration/src/test/java/org/springframework/integration/channel/DispatchingChannelErrorHandlingTests.java
View File

@@ -33,6 +33,7 @@ import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
/**
@@ -44,7 +45,7 @@ public class DispatchingChannelErrorHandlingTests {
private final CountDownLatch latch = new CountDownLatch(1);
@Test(expected = UnsupportedOperationException.class)
@Test(expected = MessageDeliveryException.class)
public void handlerThrowsExceptionPublishSubscribeWithoutExecutor() {
PublishSubscribeChannel channel = new PublishSubscribeChannel();
channel.subscribe(new MessageHandler() {

5
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/FailOverDispatcherTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2009 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.
@@ -29,7 +29,6 @@ import org.junit.Test;
import org.springframework.integration.core.Message;
import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.TestHandlers;
@@ -134,7 +133,7 @@ public class FailOverDispatcherTests {
assertEquals(6, counter.get());
}
@Test(expected = MessageDeliveryException.class)
@Test(expected = IllegalStateException.class)
public void removeConsumerLastTargetCausesDeliveryException() {
UnicastingDispatcher dispatcher = new UnicastingDispatcher();
final AtomicInteger counter = new AtomicInteger();

3
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java
View File

@@ -32,7 +32,6 @@ import org.mockito.Mock;
import org.mockito.runners.MockitoJUnit44Runner;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@@ -127,7 +126,7 @@ public class RoundRobinDispatcherConcurrentTests {
dispatcher.dispatch(message);
fail("this shouldn't happen");
}
catch (MessageDeliveryException e) {
catch (IllegalStateException e) {
// expected
}
allDone.countDown();

38
org.springframework.integration/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java
View File

@@ -25,6 +25,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
@@ -35,6 +36,7 @@ import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.StringMessage;
/**
@@ -50,6 +52,13 @@ public class DelayHandlerTests {
private final CountDownLatch latch = new CountDownLatch(1);
@Before
public void setChannelNames() {
input.setBeanName("input");
output.setBeanName("output");
}
@Test
public void noDelayHeaderAndDefaultDelayIsZero() {
DelayHandler delayHandler = new DelayHandler(0);
@@ -254,7 +263,7 @@ public class DelayHandlerTests {
assertEquals(1, latch.getCount());
}
@Test(expected = UnsupportedOperationException.class)
@Test(expected = MessageDeliveryException.class)
public void handlerThrowsExceptionWithNoDelay() {
DelayHandler delayHandler = new DelayHandler(0);
delayHandler.setOutputChannel(output);
@@ -288,10 +297,13 @@ public class DelayHandlerTests {
input.send(message);
this.waitForLatch(1000);
Message<?> errorMessage = resultHandler.lastMessage;
assertEquals(MessageDeliveryException.class, errorMessage.getPayload().getClass());
MessageDeliveryException exceptionPayload = (MessageDeliveryException) errorMessage.getPayload();
assertEquals(UnsupportedOperationException.class, exceptionPayload.getCause().getClass());
assertEquals(MessageHandlingException.class, errorMessage.getPayload().getClass());
MessageHandlingException exceptionPayload = (MessageHandlingException) errorMessage.getPayload();
assertSame(message, exceptionPayload.getFailedMessage());
assertEquals(MessageDeliveryException.class, exceptionPayload.getCause().getClass());
MessageDeliveryException nestedException = (MessageDeliveryException) exceptionPayload.getCause();
assertEquals(UnsupportedOperationException.class, nestedException.getCause().getClass());
assertSame(message, nestedException.getFailedMessage());
assertNotSame(Thread.currentThread(), resultHandler.lastThread);
}
@@ -320,10 +332,13 @@ public class DelayHandlerTests {
input.send(message);
this.waitForLatch(1000);
Message<?> errorMessage = resultHandler.lastMessage;
assertEquals(MessageDeliveryException.class, errorMessage.getPayload().getClass());
MessageDeliveryException exceptionPayload = (MessageDeliveryException) errorMessage.getPayload();
assertEquals(UnsupportedOperationException.class, exceptionPayload.getCause().getClass());
assertEquals(MessageHandlingException.class, errorMessage.getPayload().getClass());
MessageHandlingException exceptionPayload = (MessageHandlingException) errorMessage.getPayload();
assertSame(message, exceptionPayload.getFailedMessage());
assertEquals(MessageDeliveryException.class, exceptionPayload.getCause().getClass());
MessageDeliveryException nestedException = (MessageDeliveryException) exceptionPayload.getCause();
assertEquals(UnsupportedOperationException.class, nestedException.getCause().getClass());
assertSame(message, nestedException.getFailedMessage());
assertNotSame(Thread.currentThread(), resultHandler.lastThread);
}
@@ -352,10 +367,13 @@ public class DelayHandlerTests {
input.send(message);
this.waitForLatch(1000);
Message<?> errorMessage = resultHandler.lastMessage;
assertEquals(MessageDeliveryException.class, errorMessage.getPayload().getClass());
MessageDeliveryException exceptionPayload = (MessageDeliveryException) errorMessage.getPayload();
assertEquals(UnsupportedOperationException.class, exceptionPayload.getCause().getClass());
assertEquals(MessageHandlingException.class, errorMessage.getPayload().getClass());
MessageHandlingException exceptionPayload = (MessageHandlingException) errorMessage.getPayload();
assertSame(message, exceptionPayload.getFailedMessage());
assertEquals(MessageDeliveryException.class, exceptionPayload.getCause().getClass());
MessageDeliveryException nestedException = (MessageDeliveryException) exceptionPayload.getCause();
assertEquals(UnsupportedOperationException.class, nestedException.getCause().getClass());
assertSame(message, nestedException.getFailedMessage());
assertNotSame(Thread.currentThread(), resultHandler.lastThread);
}