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

INT-2431 Improve 'Dispatcher Has No Subscribers'

Browse Source 
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.
This commit is contained in:
Gary Russell
2012-02-06 19:59:41 -05:00
committed by Oleg Zhurakousky
parent 0b649abfaa
commit f75dc53ab0
14 changed files with 553 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java
View File

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

7
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java
View File

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

152
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java Normal file
View File

@@ -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<Channel>() {
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<Channel>() {
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<String> logList = insertMockLoggerInListener(amqpChannel);
MessageListener listener = (MessageListener) container.getMessageListener();
listener.onMessage(new Message("Hello world!".getBytes(), null));
verifyLogReceived(logList);
}
private List<String> insertMockLoggerInListener(
PublishSubscribeAmqpChannel channel) {
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(
channel, "container", SimpleMessageListenerContainer.class);
Log logger = mock(Log.class);
final ArrayList<String> logList = new ArrayList<String>();
doAnswer(new Answer<Object>() {
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<String> 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);
}
}