From 538b13b7574590507de60395b877ee9b75b7ce22 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 10 Dec 2009 00:13:48 +0000 Subject: [PATCH] INT-886 Failed dispatching due to "no subscribers" now provides a more meaningful exception message. --- .../channel/AbstractMessageChannel.java | 17 +++++++-- .../dispatcher/UnicastingDispatcher.java | 2 +- .../integration/handler/DelayHandler.java | 4 +- .../DispatchingChannelErrorHandlingTests.java | 3 +- .../dispatcher/FailOverDispatcherTests.java | 5 +-- .../RoundRobinDispatcherConcurrentTests.java | 3 +- .../handler/DelayHandlerTests.java | 38 ++++++++++++++----- 7 files changed, 50 insertions(+), 22 deletions(-) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 0e7db12776..7039cf428e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -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() { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java index 068f877e5c..482af6625a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java @@ -94,7 +94,7 @@ public class UnicastingDispatcher extends AbstractDispatcher { boolean success = false; Iterator handlerIterator = this.getHandlerIterator(message); if (!handlerIterator.hasNext()) { - throw new MessageDeliveryException(message, "Dispatcher has no subscribers."); + throw new IllegalStateException("Dispatcher has no subscribers."); } List exceptions = new ArrayList(); while (success == false && handlerIterator.hasNext()) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java index 45c158dbc1..0664c64484 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -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); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/DispatchingChannelErrorHandlingTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/DispatchingChannelErrorHandlingTests.java index 84d7c68772..6595bb4f75 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/DispatchingChannelErrorHandlingTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/DispatchingChannelErrorHandlingTests.java @@ -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() { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/FailOverDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/FailOverDispatcherTests.java index e3ed03fa65..78413baf73 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/FailOverDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/FailOverDispatcherTests.java @@ -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(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java index 578d3791b6..f047cbbd35 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java @@ -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(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java index 04dd2ff0d0..2eca3eb0b1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java @@ -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); }