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