From f75dc53ab07304e31c0776e696687b35a080a86e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 6 Feb 2012 19:59:41 -0500 Subject: [PATCH] INT-2431 Improve 'Dispatcher Has No Subscribers' Add channel name to MessageDeliveryException 'Dispatcher has no subscribers'. In a large integration flow, it can be difficult to track down which subscribable channel has no subscribers. This commit adds to the message text in the form for channel someChannelName to the exception message. For example: "Dispatcher has no subscribers for channel myChannel." Also for amqp-channel someAmqpChannelName for jms-channel someJMSChannelName for redis-channel someJMSChannelName INT-2431 Polishing Make component type and name immutable once set to avoid channels further down the stack frame claiming ownership. INT-2431 Polishing * PR Comments - use a new Exception type * Add support for Redis * Add WARN logs for AMQP/JMS pub-sub channels INT-2431 Polishing PR Comments: tighten up isPubSub field in AMQP-backed channel. INT-2431 Polishing Ensure channel name is not 'null' or empty string. --- .../AbstractSubscribableAmqpChannel.java | 44 ++++- .../channel/PublishSubscribeAmqpChannel.java | 7 +- .../DispatcherHasNoSubscribersTests.java | 152 ++++++++++++++++++ .../MessageDispatchingException.java | 49 ++++++ .../channel/AbstractSubscribableChannel.java | 16 +- .../dispatcher/BroadcastingDispatcher.java | 19 ++- .../dispatcher/UnicastingDispatcher.java | 5 +- ...ispatcherHasNoSubscribersTests-context.xml | 14 ++ .../DispatcherHasNoSubscribersTests.java | 66 ++++++++ .../jms/SubscribableJmsChannel.java | 42 ++++- .../jms/config/JmsChannelFactoryBean.java | 5 +- .../jms/SubscribableJmsChannelTests.java | 95 ++++++++++- .../channel/SubscribableRedisChannel.java | 19 ++- .../SubscribableRedisChannelTests.java | 52 +++++- 14 files changed, 553 insertions(+), 32 deletions(-) create mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/MessageDispatchingException.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/DispatcherHasNoSubscribersTests-context.xml create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/DispatcherHasNoSubscribersTests.java diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java index cd3f14d798..7e8eb8fc64 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -31,15 +31,19 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.beans.factory.DisposableBean; import org.springframework.context.SmartLifecycle; import org.springframework.integration.Message; +import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * @author Mark Fisher + * @author Gary Russell * @since 2.1 */ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel implements SubscribableChannel, SmartLifecycle, DisposableBean { @@ -50,16 +54,23 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple private volatile MessageDispatcher dispatcher; + private final boolean isPubSub; public AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) { + this(channelName, container, amqpTemplate, false); + } + + public AbstractSubscribableAmqpChannel(String channelName, + SimpleMessageListenerContainer container, + AmqpTemplate amqpTemplate, boolean isPubSub) { super(amqpTemplate); Assert.notNull(container, "container must not be null"); Assert.hasText(channelName, "channel name must not be empty"); this.channelName = channelName; this.container = container; + this.isPubSub = isPubSub; } - public boolean subscribe(MessageHandler handler) { return this.dispatcher.addHandler(handler); } @@ -78,7 +89,8 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple MessageConverter converter = (this.getAmqpTemplate() instanceof RabbitTemplate) ? ((RabbitTemplate) this.getAmqpTemplate()).getMessageConverter() : new SimpleMessageConverter(); - MessageListener listener = new DispatchingMessageListener(converter, this.dispatcher); + MessageListener listener = new DispatchingMessageListener(converter, + this.dispatcher, this.channelName, this.isPubSub); this.container.setMessageListener(listener); if (!this.container.isActive()) { this.container.afterPropertiesSet(); @@ -98,20 +110,27 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple private final MessageConverter converter; + private final String channelName; - private DispatchingMessageListener(MessageConverter converter, MessageDispatcher dispatcher) { + private final boolean isPubSub; + + private DispatchingMessageListener(MessageConverter converter, + MessageDispatcher dispatcher, String channelName, boolean isPubSub) { Assert.notNull(converter, "MessageConverter must not be null"); Assert.notNull(dispatcher, "MessageDispatcher must not be null"); this.converter = converter; this.dispatcher = dispatcher; + this.channelName = channelName; + this.isPubSub = isPubSub; } public void onMessage(org.springframework.amqp.core.Message message) { + Message messageToSend = null; try { Object converted = this.converter.fromMessage(message); if (converted != null) { - Message messageToSend = (converted instanceof Message) ? (Message) converted + messageToSend = (converted instanceof Message) ? (Message) converted : MessageBuilder.withPayload(converted).build(); this.dispatcher.dispatch(messageToSend); } @@ -119,6 +138,21 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple logger.warn("MessageConverter returned null, no Message to dispatch"); } } + catch (MessageDispatchingException e) { + String channelName = StringUtils.hasText(this.channelName) ? this.channelName : "unknown"; + String exceptionMessage = e.getMessage() + " for amqp-channel " + + channelName + "."; + if (this.isPubSub) { + // log only for backwards compatibility with pub/sub + if (logger.isWarnEnabled()) { + logger.warn(exceptionMessage, e); + } + } + else { + throw new MessageDeliveryException( + messageToSend, exceptionMessage, e); + } + } catch (Exception e) { throw new MessagingException("Failure occured in AMQP listener while attempting to convert and dispatch Message.", e); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java index adf5e2de6f..a3b979e70a 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -28,6 +28,7 @@ import org.springframework.integration.dispatcher.MessageDispatcher; /** * @author Mark Fisher + * @author Gary Russell * @since 2.1 */ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel { @@ -36,7 +37,7 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel public PublishSubscribeAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) { - super(channelName, container, amqpTemplate); + super(channelName, container, amqpTemplate, true); } @@ -65,7 +66,7 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel @Override protected MessageDispatcher createDispatcher() { - return new BroadcastingDispatcher(); + return new BroadcastingDispatcher(true); } @Override diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java new file mode 100644 index 0000000000..7edc89840b --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java @@ -0,0 +1,152 @@ +/* + * Copyright 2002-2012 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.amqp.channel; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyBoolean; +import static org.mockito.Matchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.junit.Test; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; +import org.springframework.amqp.core.AmqpAdmin; +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.Connection; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.test.util.TestUtils; + +import com.rabbitmq.client.Channel; + + +/** + * @author Gary Russell + * @since 2.1 + * + */ +public class DispatcherHasNoSubscribersTests { + + @Test + public void testPtP() { + final Channel channel = mock(Channel.class); + Connection connection = mock(Connection.class); + doAnswer(new Answer() { + public Channel answer(InvocationOnMock invocation) throws Throwable { + return channel; + }}).when(connection).createChannel(anyBoolean()); + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + when(connectionFactory.createConnection()).thenReturn(connection); + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + AmqpTemplate amqpTemplate = mock(AmqpTemplate.class); + + PointToPointSubscribableAmqpChannel amqpChannel = new PointToPointSubscribableAmqpChannel("noSubscribersChannel", + container, amqpTemplate); + amqpChannel.afterPropertiesSet(); + + MessageListener listener = (MessageListener) container.getMessageListener(); + try { + listener.onMessage(new Message("Hello world!".getBytes(), null)); + fail("Exception expected"); + } + catch (MessageDeliveryException e) { + assertEquals("Dispatcher has no subscribers for amqp-channel noSubscribersChannel.", e.getMessage()); + } + } + + @Test + public void testPubSub() { + final Channel channel = mock(Channel.class); + Connection connection = mock(Connection.class); + doAnswer(new Answer() { + public Channel answer(InvocationOnMock invocation) throws Throwable { + return channel; + }}).when(connection).createChannel(anyBoolean()); + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + when(connectionFactory.createConnection()).thenReturn(connection); + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + AmqpTemplate amqpTemplate = mock(AmqpTemplate.class); + final Queue queue = new Queue("noSubscribersQueue"); + PublishSubscribeAmqpChannel amqpChannel = new PublishSubscribeAmqpChannel("noSubscribersChannel", + container, amqpTemplate) { + @Override + protected Queue initializeQueue(AmqpAdmin admin, + String channelName) { + return queue; + }}; + amqpChannel.afterPropertiesSet(); + + List logList = insertMockLoggerInListener(amqpChannel); + MessageListener listener = (MessageListener) container.getMessageListener(); + listener.onMessage(new Message("Hello world!".getBytes(), null)); + verifyLogReceived(logList); + } + + private List insertMockLoggerInListener( + PublishSubscribeAmqpChannel channel) { + SimpleMessageListenerContainer container = TestUtils.getPropertyValue( + channel, "container", SimpleMessageListenerContainer.class); + Log logger = mock(Log.class); + final ArrayList logList = new ArrayList(); + doAnswer(new Answer() { + public Object answer(InvocationOnMock invocation) + throws Throwable { + String message = (String) invocation.getArguments()[0]; + if (message.startsWith("Dispatcher has no subscribers")) { + logList.add(message); + } + return null; + }}).when(logger).warn(anyString(), any(Exception.class)); + when(logger.isWarnEnabled()).thenReturn(true); + Object listener = container.getMessageListener(); + DirectFieldAccessor dfa = new DirectFieldAccessor(listener); + dfa.setPropertyValue("logger", logger); + return logList; + } + + private void verifyLogReceived(final List logList) { + assertTrue("Failed to get expected exception", logList.size() > 0); + boolean expectedExceptionFound = false; + while (logList.size() > 0) { + String message = logList.remove(0); + assertNotNull("Failed to get expected exception", message); + if (message.startsWith("Dispatcher has no subscribers")) { + expectedExceptionFound = true; + assertEquals("Dispatcher has no subscribers for amqp-channel noSubscribersChannel.", message); + break; + } + } + assertTrue("Failed to get expected exception", expectedExceptionFound); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/MessageDispatchingException.java b/spring-integration-core/src/main/java/org/springframework/integration/MessageDispatchingException.java new file mode 100644 index 0000000000..50b85945e5 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/MessageDispatchingException.java @@ -0,0 +1,49 @@ +/* + * Copyright 2002-2012 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; + +import org.springframework.integration.dispatcher.MessageDispatcher; + +/** + * Exception that indicates an internal error occurred within + * a {@link MessageDispatcher} preventing message delivery. + * + * @author Gary Russell + * @since 2.1 + * + */ +@SuppressWarnings("serial") +public class MessageDispatchingException extends MessageDeliveryException { + + public MessageDispatchingException(String description) { + super(description); + } + + public MessageDispatchingException(Message undeliveredMessage, + String description, Throwable cause) { + super(undeliveredMessage, description, cause); + } + + public MessageDispatchingException(Message undeliveredMessage, + String description) { + super(undeliveredMessage, description); + } + + public MessageDispatchingException(Message undeliveredMessage) { + super(undeliveredMessage); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java index 3c81134025..0e72ae3995 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -20,11 +20,14 @@ import java.util.concurrent.atomic.AtomicInteger; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; +import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * Base implementation of {@link MessageChannel} that invokes the subscribed @@ -32,6 +35,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel implements SubscribableChannel { @@ -58,7 +62,15 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel @Override protected boolean doSend(Message message, long timeout) { - return this.getRequiredDispatcher().dispatch(message); + try { + return this.getRequiredDispatcher().dispatch(message); + } + catch (MessageDispatchingException e) { + String componentName = this.getComponentName(); + componentName = StringUtils.hasText(componentName) ? componentName : "unknown"; + throw new MessageDeliveryException(message, e.getMessage() + + " for channel " + componentName + ".", e); + } } private MessageDispatcher getRequiredDispatcher() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java index 2cac5f6bf4..26db5535b3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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. @@ -20,6 +20,7 @@ import java.util.List; import java.util.concurrent.Executor; import org.springframework.integration.Message; +import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.support.MessageBuilder; @@ -40,6 +41,8 @@ import org.springframework.integration.support.MessageBuilder; */ public class BroadcastingDispatcher extends AbstractDispatcher { + private final boolean requireSubscribers; + private volatile boolean ignoreFailures; private volatile boolean applySequence; @@ -47,10 +50,19 @@ public class BroadcastingDispatcher extends AbstractDispatcher { private final Executor executor; public BroadcastingDispatcher() { - this.executor = null; + this(null, false); } public BroadcastingDispatcher(Executor executor) { + this(executor, false); + } + + public BroadcastingDispatcher(boolean requireSubscribers) { + this(null, requireSubscribers); + } + + public BroadcastingDispatcher(Executor executor, boolean requireSubscribers) { + this.requireSubscribers = requireSubscribers; this.executor = executor; } @@ -80,6 +92,9 @@ public class BroadcastingDispatcher extends AbstractDispatcher { boolean dispatched = false; int sequenceNumber = 1; List handlers = this.getHandlers(); + if (this.requireSubscribers && handlers.size() == 0) { + throw new MessageDispatchingException(message, "Dispatcher has no subscribers"); + } int sequenceSize = handlers.size(); for (final MessageHandler handler : handlers) { final Message messageToSend = (!this.applySequence) ? message : MessageBuilder.fromMessage(message) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java index b716984723..0c2654960a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java @@ -1,4 +1,4 @@ -/* Copyright 2002-2010 the original author or authors. +/* Copyright 2002-2012 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. @@ -25,6 +25,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock; import org.springframework.integration.Message; import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageHandler; @@ -105,7 +106,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 MessageDispatchingException(message, "Dispatcher has no subscribers"); } List exceptions = new ArrayList(); while (success == false && handlerIterator.hasNext()) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/DispatcherHasNoSubscribersTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/DispatcherHasNoSubscribersTests-context.xml new file mode 100644 index 0000000000..26228a7046 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/DispatcherHasNoSubscribersTests-context.xml @@ -0,0 +1,14 @@ + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/DispatcherHasNoSubscribersTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/DispatcherHasNoSubscribersTests.java new file mode 100644 index 0000000000..da9d488d7f --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/DispatcherHasNoSubscribersTests.java @@ -0,0 +1,66 @@ +/* + * Copyright 2002-2012 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.channel; + +import static org.junit.Assert.*; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.MessagingException; +import org.springframework.integration.message.GenericMessage; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 2.1 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class DispatcherHasNoSubscribersTests { + + @Autowired + MessageChannel noSubscribersChannel; + + @Autowired + MessageChannel subscribedChannel; + + @Test + public void oneChannel() { + try { + noSubscribersChannel.send(new GenericMessage("Hello, world!")); + fail("Exception expected"); + } catch (MessagingException e) { + assertEquals("Dispatcher has no subscribers for channel noSubscribersChannel.", e.getMessage()); + assertEquals("Dispatcher has no subscribers for channel noSubscribersChannel.", e.getLocalizedMessage()); + } + } + + @Test + public void stackedChannels() { + try { + subscribedChannel.send(new GenericMessage("Hello, world!")); + fail("Exception expected"); + } catch (MessagingException e) { + assertEquals("Dispatcher has no subscribers for channel noSubscribersChannel.", e.getMessage()); + assertEquals("Dispatcher has no subscribers for channel noSubscribersChannel.", e.getLocalizedMessage()); + } + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java index e14c140341..6536016497 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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. @@ -24,6 +24,8 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.DisposableBean; import org.springframework.context.SmartLifecycle; import org.springframework.integration.Message; +import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; @@ -35,9 +37,11 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.listener.AbstractMessageListenerContainer; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * @author Mark Fisher + * @author Gary Russell * @since 2.0 */ public class SubscribableJmsChannel extends AbstractJmsChannel implements SubscribableChannel, SmartLifecycle, DisposableBean { @@ -71,8 +75,11 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr return; } super.onInit(); - this.configureDispatcher(this.container.isPubSubDomain()); - MessageListener listener = new DispatchingMessageListener(this.getJmsTemplate(), this.dispatcher); + boolean isPubSub = this.container.isPubSubDomain(); + this.configureDispatcher(isPubSub); + MessageListener listener = new DispatchingMessageListener( + this.getJmsTemplate(), this.dispatcher, + this.getComponentName(), isPubSub); this.container.setMessageListener(listener); if (!this.container.isActive()) { this.container.afterPropertiesSet(); @@ -82,7 +89,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr private void configureDispatcher(boolean isPubSub) { if (isPubSub) { - this.dispatcher = new BroadcastingDispatcher(); + this.dispatcher = new BroadcastingDispatcher(true); } else { UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(); @@ -100,18 +107,26 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr private final MessageDispatcher dispatcher; + private final String channelName; - private DispatchingMessageListener(JmsTemplate jmsTemplate, MessageDispatcher dispatcher) { + private final boolean isPubSub; + + + private DispatchingMessageListener(JmsTemplate jmsTemplate, + MessageDispatcher dispatcher, String channelName, boolean isPubSub) { this.jmsTemplate = jmsTemplate; this.dispatcher = dispatcher; + this.channelName = channelName; + this.isPubSub = isPubSub; } public void onMessage(javax.jms.Message message) { + Message messageToSend = null; try { Object converted = this.jmsTemplate.getMessageConverter().fromMessage(message); if (converted != null) { - Message messageToSend = (converted instanceof Message) ? (Message) converted + messageToSend = (converted instanceof Message) ? (Message) converted : MessageBuilder.withPayload(converted).build(); this.dispatcher.dispatch(messageToSend); } @@ -119,6 +134,21 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr logger.warn("MessageConverter returned null, no Message to dispatch"); } } + catch (MessageDispatchingException e) { + String channelName = StringUtils.hasText(this.channelName) ? this.channelName : "unknown"; + String exceptionMessage = e.getMessage() + " for jms-channel " + + channelName + "."; + if (this.isPubSub) { + // log only for backwards compatibility with pub/sub + if (logger.isWarnEnabled()) { + logger.warn(exceptionMessage, e); + } + } + else { + throw new MessageDeliveryException( + messageToSend, exceptionMessage, e); + } + } catch (Exception e) { throw new MessagingException("failed to handle incoming JMS Message", e); } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java index 7c13f335c1..88f3b8bbd4 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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. @@ -48,6 +48,7 @@ import org.springframework.util.StringUtils; /** * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.0 */ public class JmsChannelFactoryBean extends AbstractFactoryBean implements SmartLifecycle, DisposableBean, BeanNameAware { @@ -337,8 +338,8 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean logList = insertMockLoggerInListener(channel); + listener.onMessage(new StubTextMessage("Hello, world!")); + verifyLogReceived(logList); + } + + private List insertMockLoggerInListener( + SubscribableJmsChannel channel) { + AbstractMessageListenerContainer container = TestUtils.getPropertyValue( + channel, "container", AbstractMessageListenerContainer.class); + Log logger = mock(Log.class); + final ArrayList logList = new ArrayList(); + doAnswer(new Answer() { + public Object answer(InvocationOnMock invocation) + throws Throwable { + String message = (String) invocation.getArguments()[0]; + if (message.startsWith("Dispatcher has no subscribers")) { + logList.add(message); + } + return null; + }}).when(logger).warn(anyString(), any(Exception.class)); + when(logger.isWarnEnabled()).thenReturn(true); + Object listener = container.getMessageListener(); + DirectFieldAccessor dfa = new DirectFieldAccessor(listener); + dfa.setPropertyValue("logger", logger); + return logList; + } + + private void verifyLogReceived(final List logList) { + assertTrue("Failed to get expected exception", logList.size() > 0); + boolean expectedExceptionFound = false; + while (logList.size() > 0) { + String message = logList.remove(0); + assertNotNull("Failed to get expected exception", message); + if (message.startsWith("Dispatcher has no subscribers")) { + expectedExceptionFound = true; + assertEquals("Dispatcher has no subscribers for jms-channel noSubscribersChannel.", message); + break; + } + } + assertTrue("Failed to get expected exception", expectedExceptionFound); + } /** * Blocks until the listener container has subscribed; if the container does not support diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java index fe0f3e53d7..13bd1b2cd2 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -30,6 +30,8 @@ import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.integration.Message; +import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.core.MessageHandler; @@ -42,9 +44,11 @@ import org.springframework.integration.support.converter.SimpleMessageConverter; import org.springframework.integration.util.ErrorHandlingTaskExecutor; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; +import org.springframework.util.StringUtils; /** * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.0 */ @SuppressWarnings("rawtypes") @@ -55,7 +59,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements private final RedisTemplate redisTemplate; private final String topicName; - private final MessageDispatcher dispatcher = new BroadcastingDispatcher(); + private final MessageDispatcher dispatcher = new BroadcastingDispatcher(true); private volatile boolean initialized; @@ -171,7 +175,16 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements @SuppressWarnings("unused") public void handleMessage(String s) { Message siMessage = messageConverter.toMessage(s); - dispatcher.dispatch(siMessage); + try { + dispatcher.dispatch(siMessage); + } + catch (MessageDispatchingException e) { + String topicName = SubscribableRedisChannel.this.topicName; + topicName = StringUtils.hasText(topicName) ? topicName : "unknown"; + throw new MessageDeliveryException(siMessage, e.getMessage() + + " for redis-channel " + + topicName + ".", e); + } } } } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java index bf2705dadf..cd796db7e8 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -15,21 +15,33 @@ */ package org.springframework.integration.redis.channel; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import java.lang.reflect.InvocationTargetException; +import java.util.Map; +import java.util.Set; + import org.junit.Test; import org.mockito.Mockito; import org.springframework.beans.factory.BeanFactory; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; +import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; import org.springframework.integration.Message; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; - -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.util.ReflectionUtils; /** * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.0 */ public class SubscribableRedisChannelTests extends RedisAvailableTests{ @@ -56,4 +68,34 @@ public class SubscribableRedisChannelTests extends RedisAvailableTests{ verify(handler, times(3)).handleMessage(Mockito.any(Message.class)); channel.stop(); } + + @Test + @RedisAvailable + public void dispatcherHasNoSubscribersTest() throws Exception{ + JedisConnectionFactory connectionFactory = new JedisConnectionFactory(); + connectionFactory.setPort(7379); + connectionFactory.afterPropertiesSet(); + + SubscribableRedisChannel channel = new SubscribableRedisChannel(connectionFactory, "si.test.channel.no.subs"); + channel.setBeanFactory(mock(BeanFactory.class)); + channel.afterPropertiesSet(); + + RedisMessageListenerContainer container = TestUtils.getPropertyValue( + channel, "container", RedisMessageListenerContainer.class); + @SuppressWarnings("unchecked") + Map> channelMapping = (Map>) TestUtils + .getPropertyValue(container, "channelMapping"); + MessageListenerAdapter listener = channelMapping.entrySet().iterator().next().getValue().iterator().next(); + Object delegate = TestUtils.getPropertyValue(listener, "delegate"); + try { + ReflectionUtils.findMethod(delegate.getClass(), "handleMessage", String.class).invoke(delegate, "Hello, world!"); + fail("Exception expected"); + } + catch (InvocationTargetException e) { + Throwable cause = e.getCause(); + assertNotNull(cause); + assertEquals("Dispatcher has no subscribers for redis-channel si.test.channel.no.subs.", cause.getMessage()); + } + + } }