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

INT-4371: More polishing - consumer queue header

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4371
This commit is contained in:
Gary Russell
2018-01-11 15:02:59 -05:00
committed by Artem Bilan
parent c98d50540c
commit 00b910c874
2 changed files with 4 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java
View File

@@ -169,6 +169,7 @@ public class AmqpMessageSource extends AbstractMessageSource<Object> {
.createCallback(new AmqpAckInfo(connection, channel, this.transacted, resp));
MessageProperties messageProperties = this.propertiesConverter.toMessageProperties(resp.getProps(),
resp.getEnvelope(), StandardCharsets.UTF_8.name());
messageProperties.setConsumerQueue(this.queue);
Map<String, Object> headers = this.headerMapper.toHeadersFromRequest(messageProperties);
org.springframework.amqp.core.Message amqpMessage = new org.springframework.amqp.core.Message(resp.getBody(), messageProperties);
Object payload = this.messageConverter.fromMessage(amqpMessage);

3
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.amqp.inbound;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertThat;
import static org.mockito.ArgumentMatchers.anyString;
@@ -32,6 +33,7 @@ import java.util.concurrent.TimeoutException;
import org.junit.Test;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
import org.springframework.integration.support.AcknowledgmentCallback.Status;
import org.springframework.integration.support.StaticMessageHeaderAccessor;
@@ -72,6 +74,7 @@ public class AmqpMessageSourceTests {
Message<?> received = source.receive();
assertThat(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE),
instanceOf(org.springframework.amqp.core.Message.class));
assertThat(received.getHeaders().get(AmqpHeaders.CONSUMER_QUEUE), equalTo("foo"));
// make sure channel is not cached
org.springframework.amqp.rabbit.connection.Connection conn = ccf.createConnection();
Channel notCached = conn.createChannel(false); // should not have been "closed"