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

INT-3602: Fix AMQP Direct ReplyTo Handling

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

Direct reply-to addresses can contain '/' (base64); need to use AddressUtils to
properly decode reply-to addresses.

Also suppress a test error in the Spring IO compatibility build until
Spring IO updates to spring-amqp 1.4.2.
This commit is contained in:
Gary Russell
2015-01-10 12:52:47 -05:00
parent 1ac1c734b3
commit 48ac9a16be
2 changed files with 18 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -18,11 +18,10 @@ package org.springframework.integration.amqp.inbound;
import java.util.Map;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Address;
import org.springframework.amqp.core.AddressUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
@@ -38,6 +37,8 @@ import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import com.rabbitmq.client.Channel;
/**
* Adapter that receives Messages from an AMQP Queue, converts them into
* Spring Integration Messages, and sends the results to a Message Channel.
@@ -103,7 +104,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
final org.springframework.messaging.Message<?> reply = sendAndReceiveMessage(request);
if (reply != null) {
// TODO: fallback to a reply address property of this gateway
Address replyTo = message.getMessageProperties().getReplyToAddress();
Address replyTo = AddressUtils.decodeReplyToAddress(message);
Assert.notNull(replyTo, "The replyTo header must not be null on a " +
"request Message being handled by the AMQP inbound gateway.");
amqpTemplate.convertAndSend(replyTo.getExchangeName(), replyTo.getRoutingKey(), reply.getPayload(),

21
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -173,13 +173,18 @@ public class DefaultAmqpHeaderMapperTests {
@Test // INT-3586 requires Spring AMQP 1.4.2
public void testToHeadersConsumerMetadata() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setConsumerTag("consumerTag");
amqpProperties.setConsumerQueue("consumerQueue");
Map<String, Object> headerMap = headerMapper.toHeadersFromRequest(amqpProperties);
assertEquals("consumerTag", headerMap.get(AmqpHeaders.CONSUMER_TAG));
assertEquals("consumerQueue", headerMap.get(AmqpHeaders.CONSUMER_QUEUE));
try {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setConsumerTag("consumerTag");
amqpProperties.setConsumerQueue("consumerQueue");
Map<String, Object> headerMap = headerMapper.toHeadersFromRequest(amqpProperties);
assertEquals("consumerTag", headerMap.get(AmqpHeaders.CONSUMER_TAG));
assertEquals("consumerQueue", headerMap.get(AmqpHeaders.CONSUMER_QUEUE));
}
catch (NoSuchMethodError e) {
// TODO: temporary for Spring IO Compatibility build - until Spring IO 1.1.x moves to amqp 1.4.2
}
}
@Test