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

INT-4256: AMQP: Conversion Errors to ErrorChannel

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4256

Also fix the JMS endpoint to use the `MessagingTemplate` instead of sending to the
error channel directly (ignored the send result).

Missing commit

Polishing - PR Comments

* Some additional polishing: remove extra `ifs`; make internal
classes `protected` for possible inheritors

Conflicts:
	spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
	spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java

AMQP: Add Support for SCSt Error Handling

See: https://github.com/spring-cloud/spring-cloud-stream/issues/913

Add retry within `onMessage` so we have access to the converted message
as well as the original spring-amqp message, which is added to the
`ErrorMessage` as a header.

Similar to the work in spring-integration-kafka.

* Polishing JavaDocs
* Rename `AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_DATA`
to `AmqpMessageHeaderErrorMessageStrategy.AMQP_MESSAGE`
This commit is contained in:
Gary Russell
2017-05-22 10:52:14 -04:00
committed by Artem Bilan
parent 37cf8336de
commit 738d0b0757
5 changed files with 530 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -22,13 +22,23 @@ import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
import com.rabbitmq.client.Channel;
@@ -40,17 +50,23 @@ import com.rabbitmq.client.Channel;
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.1
*/
public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
OrderlyShutdownCapable {
private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<AttributeAccessor>();
private final AbstractMessageListenerContainer messageListenerContainer;
private volatile MessageConverter messageConverter = new SimpleMessageConverter();
private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
private RetryTemplate retryTemplate;
private RecoveryCallback<? extends Object> recoveryCallback;
public AmqpInboundChannelAdapter(AbstractMessageListenerContainer listenerContainer) {
Assert.notNull(listenerContainer, "listenerContainer must not be null");
@@ -60,6 +76,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
"configure its own listener implementation.");
this.messageListenerContainer = listenerContainer;
this.messageListenerContainer.setAutoStartup(false);
setErrorMessageStrategy(new AmqpMessageHeaderErrorMessageStrategy());
}
@@ -73,6 +90,31 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
this.headerMapper = headerMapper;
}
/**
* Set a {@link RetryTemplate} to use for retrying a message delivery within the
* adapter. Unlike adding retry at the container level, this can be used with an
* {@code ErrorMessageSendingRecoverer} {@link RecoveryCallback} to publish to the
* error channel after retries are exhausted. You generally should not configure an
* error channel when using retry here, use a {@link RecoveryCallback} instead.
* @param retryTemplate the template.
* @since 4.3.10.
* @see #setRecoveryCallback(RecoveryCallback)
*/
public void setRetryTemplate(RetryTemplate retryTemplate) {
this.retryTemplate = retryTemplate;
}
/**
* Set a {@link RecoveryCallback} when using retry within the adapter.
* @param recoveryCallback the callback.
* @since 4.3.10
* @see #setRetryTemplate(RetryTemplate)
*/
public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
this.recoveryCallback = recoveryCallback;
}
@Override
public String getComponentType() {
return "amqp:inbound-channel-adapter";
@@ -80,22 +122,16 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
@Override
protected void onInit() {
this.messageListenerContainer.setMessageListener(new ChannelAwareMessageListener() {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
Object payload = AmqpInboundChannelAdapter.this.messageConverter.fromMessage(message);
Map<String, Object> headers =
AmqpInboundChannelAdapter.this.headerMapper.toHeadersFromRequest(message.getMessageProperties());
if (AmqpInboundChannelAdapter.this.messageListenerContainer.getAcknowledgeMode()
== AcknowledgeMode.MANUAL) {
headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag());
headers.put(AmqpHeaders.CHANNEL, channel);
}
sendMessage(getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build());
}
});
if (this.retryTemplate != null) {
Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
+ "send an error message when retries are exhausted");
}
Listener messageListener = new Listener();
if (this.retryTemplate != null) {
this.retryTemplate.registerListener(messageListener);
}
this.messageListenerContainer.setMessageListener(messageListener);
this.messageListenerContainer.afterPropertiesSet();
super.onInit();
}
@@ -110,26 +146,122 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
this.messageListenerContainer.stop();
}
/**
* {@inheritDoc}
* <p>
* Shuts down the listener container.
*/
@Override
public int beforeShutdown() {
this.stop();
return 0;
}
/**
* {@inheritDoc}
* <p>No-op
*/
@Override
public int afterShutdown() {
return 0;
}
/**
* If there's a retry template, it will set the attributes holder via the listener. If
* there's no retry template, but there's an error channel, we create a new attributes
* holder here. If an attributes holder exists (by either method), we set the
* attributes for use by the {@link ErrorMessageStrategy}.
* @param amqpMessage the AMQP message to use.
* @param message the Spring Messaging message to use.
* @since 4.3.10
*/
private void setAttributesIfNecessary(Message amqpMessage, org.springframework.messaging.Message<?> message) {
boolean needHolder = getErrorChannel() != null && this.retryTemplate == null;
boolean needAttributes = needHolder || this.retryTemplate != null;
if (needHolder) {
attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
}
if (needAttributes) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes != null) {
attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
attributes.setAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_MESSAGE, amqpMessage);
}
}
}
@Override
protected AttributeAccessor getErrorMessageAttributes(org.springframework.messaging.Message<?> message) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes == null) {
return super.getErrorMessageAttributes(message);
}
else {
return attributes;
}
}
protected class Listener implements ChannelAwareMessageListener, RetryListener {
@SuppressWarnings("unchecked")
@Override
public void onMessage(final Message message, final Channel channel) throws Exception {
try {
if (AmqpInboundChannelAdapter.this.retryTemplate == null) {
processMessage(message, channel);
}
else {
AmqpInboundChannelAdapter.this.retryTemplate.execute(new RetryCallback<Object, RuntimeException>() {
@Override
public Void doWithRetry(RetryContext context) throws RuntimeException {
processMessage(message, channel);
return null;
}
}, (RecoveryCallback<Object>) AmqpInboundChannelAdapter.this.recoveryCallback);
}
}
catch (RuntimeException e) {
if (getErrorChannel() != null) {
getMessagingTemplate().send(getErrorChannel(), buildErrorMessage(null,
new ListenerExecutionFailedException("Message conversion failed", e, message)));
}
else {
throw e;
}
}
finally {
attributesHolder.remove();
}
}
private void processMessage(Message message, Channel channel) {
Object payload = AmqpInboundChannelAdapter.this.messageConverter.fromMessage(message);
Map<String, Object> headers = AmqpInboundChannelAdapter.this.headerMapper
.toHeadersFromRequest(message.getMessageProperties());
if (AmqpInboundChannelAdapter.this.messageListenerContainer.getAcknowledgeMode()
== AcknowledgeMode.MANUAL) {
headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag());
headers.put(AmqpHeaders.CHANNEL, channel);
}
final org.springframework.messaging.Message<Object> messagingMessage = getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(headers)
.build();
setAttributesIfNecessary(message, messagingMessage);
sendMessage(messagingMessage);
}
@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
attributesHolder.set(context);
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
}
}

204
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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,12 +28,22 @@ import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -47,10 +57,14 @@ import com.rabbitmq.client.Channel;
*
* @author Mark Fisher
* @author Artem Bilan
* @author Gary Russell
*
* @since 2.1
*/
public class AmqpInboundGateway extends MessagingGatewaySupport {
private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<AttributeAccessor>();
private final AbstractMessageListenerContainer messageListenerContainer;
private final AmqpTemplate amqpTemplate;
@@ -63,6 +77,10 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
private Address defaultReplyTo;
private RetryTemplate retryTemplate;
private RecoveryCallback<? extends Object> recoveryCallback;
public AmqpInboundGateway(AbstractMessageListenerContainer listenerContainer) {
this(listenerContainer, new RabbitTemplate(listenerContainer.getConnectionFactory()), false);
}
@@ -90,6 +108,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
this.messageListenerContainer.setAutoStartup(false);
this.amqpTemplate = amqpTemplate;
this.amqpTemplateExplicitlySet = amqpTemplateExplicitlySet;
setErrorMessageStrategy(new AmqpMessageHeaderErrorMessageStrategy());
}
@@ -132,6 +151,30 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
this.defaultReplyTo = new Address(defaultReplyTo);
}
/**
* Set a {@link RetryTemplate} to use for retrying a message delivery within the
* gateway. Unlike adding retry at the container level, this can be used with an
* {@code ErrorMessageSendingRecoverer} {@link RecoveryCallback} to publish to the
* error channel after retries are exhausted. You generally should not configure an
* error channel when using retry here, use a {@link RecoveryCallback} instead.
* @param retryTemplate the template.
* @since 4.3.10.
* @see #setRecoveryCallback(RecoveryCallback)
*/
public void setRetryTemplate(RetryTemplate retryTemplate) {
this.retryTemplate = retryTemplate;
}
/**
* Set a {@link RecoveryCallback} when using retry within the gateway.
* @param recoveryCallback the callback.
* @since 4.3.10
* @see #setRetryTemplate(RetryTemplate)
*/
public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
this.recoveryCallback = recoveryCallback;
}
@Override
public String getComponentType() {
return "amqp:inbound-gateway";
@@ -139,19 +182,124 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
@Override
protected void onInit() throws Exception {
this.messageListenerContainer.setMessageListener(new ChannelAwareMessageListener() {
@Override
public void onMessage(Message message, Channel channel) {
Object payload = AmqpInboundGateway.this.amqpMessageConverter.fromMessage(message);
Map<String, Object> headers =
AmqpInboundGateway.this.headerMapper.toHeadersFromRequest(message.getMessageProperties());
if (this.retryTemplate != null) {
Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
+ "send an error message when retries are exhausted");
}
Listener messageListener = new Listener();
if (this.retryTemplate != null) {
this.retryTemplate.registerListener(messageListener);
}
this.messageListenerContainer.setMessageListener(messageListener);
this.messageListenerContainer.afterPropertiesSet();
if (!this.amqpTemplateExplicitlySet) {
((RabbitTemplate) this.amqpTemplate).afterPropertiesSet();
}
super.onInit();
if (this.retryTemplate != null && getErrorChannel() != null) {
logger.warn("Usually, when using a RetryTemplate you should use an ErrorMessageSendingRecoverer and not "
+ "provide an errorChannel. Using an errorChannel could defeat retry and will receive an error "
+ "message for each delivery attempt.");
}
}
@Override
protected void doStart() {
this.messageListenerContainer.start();
}
@Override
protected void doStop() {
this.messageListenerContainer.stop();
}
/**
* If there's a retry template, it will set the attributes holder via the listener. If
* there's no retry template, but there's an error channel, we create a new attributes
* holder here. If an attributes holder exists (by either method), we set the
* attributes for use by the {@link ErrorMessageStrategy}.
* @param amqpMessage the AMQP message to use.
* @param message the Spring Messaging message to use.
* @since 4.3.10
*/
private void setAttributesIfNecessary(Message amqpMessage, org.springframework.messaging.Message<?> message) {
boolean needHolder = getErrorChannel() != null && this.retryTemplate == null;
boolean needAttributes = needHolder || this.retryTemplate != null;
if (needHolder) {
attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
}
if (needAttributes) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes != null) {
attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
attributes.setAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_MESSAGE, amqpMessage);
}
}
}
@Override
protected AttributeAccessor getErrorMessageAttributes(org.springframework.messaging.Message<?> message) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes == null) {
return super.getErrorMessageAttributes(message);
}
else {
return attributes;
}
}
protected class Listener implements ChannelAwareMessageListener, RetryListener {
@SuppressWarnings("unchecked")
@Override
public void onMessage(final Message message, final Channel channel) throws Exception {
if (AmqpInboundGateway.this.retryTemplate == null) {
doOnMessage(message, channel);
}
else {
AmqpInboundGateway.this.retryTemplate.execute(new RetryCallback<Object, RuntimeException>() {
@Override
public Object doWithRetry(RetryContext context) throws RuntimeException {
doOnMessage(message, channel);
return null;
}
}, (RecoveryCallback<Object>) AmqpInboundGateway.this.recoveryCallback);
}
}
private void doOnMessage(Message message, Channel channel) {
boolean error = false;
Map<String, Object> headers = null;
Object payload = null;
try {
payload = AmqpInboundGateway.this.amqpMessageConverter.fromMessage(message);
headers = AmqpInboundGateway.this.headerMapper.toHeadersFromRequest(message.getMessageProperties());
if (AmqpInboundGateway.this.messageListenerContainer.getAcknowledgeMode() == AcknowledgeMode.MANUAL) {
headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag());
headers.put(AmqpHeaders.CHANNEL, channel);
}
org.springframework.messaging.Message<?> request =
getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build();
final org.springframework.messaging.Message<?> reply = sendAndReceiveMessage(request);
}
catch (RuntimeException e) {
if (getErrorChannel() != null) {
AmqpInboundGateway.this.messagingTemplate.send(getErrorChannel(), buildErrorMessage(null,
new ListenerExecutionFailedException("Message conversion failed", e, message)));
}
else {
throw e;
}
error = true;
}
if (!error) {
org.springframework.messaging.Message<Object> messagingMessage = getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(headers)
.build();
setAttributesIfNecessary(message, messagingMessage);
final org.springframework.messaging.Message<?> reply = sendAndReceiveMessage(messagingMessage);
if (reply != null) {
Address replyTo;
String replyToProperty = message.getMessageProperties().getReplyTo();
@@ -170,8 +318,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
String contentEncoding = messageProperties.getContentEncoding();
long contentLength = messageProperties.getContentLength();
String contentType = messageProperties.getContentType();
AmqpInboundGateway.this.headerMapper.fromHeadersToReply(reply.getHeaders(),
messageProperties);
AmqpInboundGateway.this.headerMapper.fromHeadersToReply(reply.getHeaders(), messageProperties);
// clear the replyTo from the original message since we are using it now
messageProperties.setReplyTo(null);
// reset the content-* properties as determined by the MessageConverter
@@ -183,7 +330,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
messageProperties.setContentType(contentType);
}
return message;
}
};
};
@@ -203,23 +350,26 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
}
}
}
});
this.messageListenerContainer.afterPropertiesSet();
if (!this.amqpTemplateExplicitlySet) {
((RabbitTemplate) this.amqpTemplate).afterPropertiesSet();
}
super.onInit();
}
@Override
protected void doStart() {
this.messageListenerContainer.start();
}
@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
attributesHolder.set(context);
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
@Override
protected void doStop() {
this.messageListenerContainer.stop();
}
}

56
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java Normal file
View File

@@ -0,0 +1,56 @@
/*
* Copyright 2017 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.support;
import java.util.Collections;
import java.util.Map;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
/**
* An {@link ErrorMessageStrategy} extension that adds the raw AMQP message as
* a header to the {@code org.springframework.integration.message.EnhancedErrorMessage}.
*
* @author Gary Russell
* @since 4.3.10
*
*/
public class AmqpMessageHeaderErrorMessageStrategy implements ErrorMessageStrategy {
/**
* Header name/retry context variable for the raw received message.
*/
public static final String AMQP_MESSAGE = "amqp_message"; // move to AmqpHeaders.MESSAGE in 2.0
@SuppressWarnings("deprecation")
@Override
public ErrorMessage buildErrorMessage(Throwable throwable, AttributeAccessor context) {
Object inputMessage = context.getAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY);
Map<String, Object> headers = Collections.singletonMap(
AmqpMessageHeaderErrorMessageStrategy.AMQP_MESSAGE,
context.getAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_MESSAGE));
return inputMessage instanceof Message
? new org.springframework.integration.message.EnhancedErrorMessage(throwable, headers,
(Message<?>) inputMessage)
: new ErrorMessage(throwable, headers);
}
}

128
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
View File

@@ -16,10 +16,16 @@
package org.springframework.integration.amqp.inbound;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.anyBoolean;
import static org.mockito.Mockito.doAnswer;
@@ -39,15 +45,20 @@ import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
import org.springframework.integration.json.JsonToObjectTransformer;
import org.springframework.integration.json.ObjectToJsonTransformer;
import org.springframework.integration.mapping.support.JsonHeaders;
@@ -55,8 +66,10 @@ import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.transformer.MessageTransformingHandler;
import org.springframework.integration.transformer.Transformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.retry.support.RetryTemplate;
import com.rabbitmq.client.Channel;
@@ -225,6 +238,121 @@ public class InboundEndpointTests {
}
@Test
public void testAdapterConversionError() throws Exception {
Connection connection = mock(Connection.class);
doAnswer(invocation -> mock(Channel.class)).when(connection).createChannel(anyBoolean());
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
when(connectionFactory.createConnection()).thenReturn(connection);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container);
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
QueueChannel errorChannel = new QueueChannel();
adapter.setErrorChannel(errorChannel);
adapter.setMessageConverter(new MessageConverter() {
@Override
public org.springframework.amqp.core.Message toMessage(Object object, MessageProperties messageProperties)
throws MessageConversionException {
throw new MessageConversionException("intended");
}
@Override
public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException {
return null;
}
});
adapter.afterPropertiesSet();
((ChannelAwareMessageListener) container.getMessageListener()).onMessage(null, null);
assertNull(outputChannel.receive(0));
assertNotNull(errorChannel.receive(0));
}
@Test
public void testGatewayConversionError() throws Exception {
Connection connection = mock(Connection.class);
doAnswer(invocation -> mock(Channel.class)).when(connection).createChannel(anyBoolean());
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
when(connectionFactory.createConnection()).thenReturn(connection);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
AmqpInboundGateway adapter = new AmqpInboundGateway(container);
QueueChannel outputChannel = new QueueChannel();
adapter.setRequestChannel(outputChannel);
QueueChannel errorChannel = new QueueChannel();
adapter.setErrorChannel(errorChannel);
adapter.setMessageConverter(new MessageConverter() {
@Override
public org.springframework.amqp.core.Message toMessage(Object object, MessageProperties messageProperties)
throws MessageConversionException {
throw new MessageConversionException("intended");
}
@Override
public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException {
return null;
}
});
adapter.afterPropertiesSet();
((ChannelAwareMessageListener) container.getMessageListener()).onMessage(null, null);
assertNull(outputChannel.receive(0));
assertNotNull(errorChannel.receive(0));
}
@Test
public void testRetryWithinOnMessageAdapter() throws Exception {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
AbstractMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container);
adapter.setOutputChannel(new DirectChannel());
adapter.setRetryTemplate(new RetryTemplate());
QueueChannel errors = new QueueChannel();
ErrorMessageSendingRecoverer recoveryCallback = new ErrorMessageSendingRecoverer(errors);
recoveryCallback.setErrorMessageStrategy(new AmqpMessageHeaderErrorMessageStrategy());
adapter.setRecoveryCallback(recoveryCallback);
adapter.afterPropertiesSet();
ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener();
listener.onMessage(org.springframework.amqp.core.MessageBuilder.withBody("foo".getBytes())
.andProperties(new MessageProperties()).build(), null);
Message<?> errorMessage = errors.receive(0);
assertNotNull(errorMessage);
assertThat(errorMessage.getPayload(), instanceOf(MessagingException.class));
assertThat(((MessagingException) errorMessage.getPayload()).getMessage(), containsString("Dispatcher has no"));
org.springframework.amqp.core.Message amqpMessage = errorMessage.getHeaders()
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_MESSAGE, org.springframework.amqp.core.Message.class);
assertThat(amqpMessage, notNullValue());
assertNull(errors.receive(0));
}
@Test
public void testRetryWithinOnMessageGateway() throws Exception {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
AbstractMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
AmqpInboundGateway adapter = new AmqpInboundGateway(container);
adapter.setRequestChannel(new DirectChannel());
adapter.setRetryTemplate(new RetryTemplate());
QueueChannel errors = new QueueChannel();
ErrorMessageSendingRecoverer recoveryCallback = new ErrorMessageSendingRecoverer(errors);
recoveryCallback.setErrorMessageStrategy(new AmqpMessageHeaderErrorMessageStrategy());
adapter.setRecoveryCallback(recoveryCallback);
adapter.afterPropertiesSet();
ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener();
listener.onMessage(org.springframework.amqp.core.MessageBuilder.withBody("foo".getBytes())
.andProperties(new MessageProperties()).build(), null);
Message<?> errorMessage = errors.receive(0);
assertNotNull(errorMessage);
assertThat(errorMessage.getPayload(), instanceOf(MessagingException.class));
assertThat(((MessagingException) errorMessage.getPayload()).getMessage(), containsString("Dispatcher has no"));
org.springframework.amqp.core.Message amqpMessage = errorMessage.getHeaders()
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_MESSAGE, org.springframework.amqp.core.Message.class);
assertThat(amqpMessage, notNullValue());
assertNull(errors.receive(0));
}
public static class Foo {

12
spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java
View File

@@ -32,6 +32,7 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
@@ -330,8 +331,9 @@ public class ChannelPublishingJmsMessageListener
if (errorChannel == null) {
throw e;
}
errorChannel.send(this.gatewayDelegate.buildErrorMessage(
new MessagingException("Inbound conversion failed for: " + jmsMessage, e)));
this.gatewayDelegate.getMessagingTemplate().send(errorChannel,
this.gatewayDelegate.buildErrorMessage(
new MessagingException("Inbound conversion failed for: " + jmsMessage, e)));
errors = true;
}
if (!errors) {
@@ -510,10 +512,14 @@ public class ChannelPublishingJmsMessageListener
return super.sendAndReceiveMessage(request);
}
public ErrorMessage buildErrorMessage(Throwable throwable) {
protected ErrorMessage buildErrorMessage(Throwable throwable) {
return super.buildErrorMessage(null, throwable);
}
protected MessagingTemplate getMessagingTemplate() {
return this.messagingTemplate;
}
@Override
public String getComponentType() {
if (ChannelPublishingJmsMessageListener.this.expectReply) {