From be90fdb1de9462543a56cd02bfde30b7b57c133f Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 10 Jun 2019 17:20:31 -0400 Subject: [PATCH] AMQP: Add bindSourceMessage property (inbound) Resolves https://github.com/spring-projects/spring-integration/issues/2958 # 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 # spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceTests.java # spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java # spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java --- .../inbound/AmqpInboundChannelAdapter.java | 15 +++++++++++++++ .../amqp/inbound/AmqpInboundGateway.java | 15 +++++++++++++++ .../amqp/inbound/AmqpMessageSource.java | 10 ++++++---- .../amqp/inbound/AmqpMessageSourceTests.java | 5 +++++ .../amqp/inbound/InboundEndpointTests.java | 5 +++++ .../IntegrationMessageHeaderAccessor.java | 18 ++++++++++++++++++ .../StaticMessageHeaderAccessor.java | 6 ++++++ 7 files changed, 70 insertions(+), 4 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index da15dc643a..fd3a581757 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java @@ -69,6 +69,8 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements private RecoveryCallback recoveryCallback; + private boolean bindSourceMessage; + public AmqpInboundChannelAdapter(AbstractMessageListenerContainer listenerContainer) { Assert.notNull(listenerContainer, "listenerContainer must not be null"); Assert.isNull(listenerContainer.getMessageListener(), @@ -116,6 +118,16 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements } + /** + * Set to true to bind the source message in the header named + * {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}. + * @param bindSourceMessage true to bind. + * @since 5.1.6 + */ + public void setBindSourceMessage(boolean bindSourceMessage) { + this.bindSourceMessage = bindSourceMessage; + } + @Override public String getComponentType() { return "amqp:inbound-channel-adapter"; @@ -249,6 +261,9 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements if (AmqpInboundChannelAdapter.this.retryTemplate != null) { headers.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, new AtomicInteger()); } + if (AmqpInboundChannelAdapter.this.bindSourceMessage) { + headers.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, message); + } final org.springframework.messaging.Message messagingMessage = getMessageBuilderFactory() .withPayload(payload) .copyHeaders(headers) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java index 92c9a55408..6c10153eed 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java @@ -81,6 +81,8 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { private RecoveryCallback recoveryCallback; + private boolean bindSourceMessage; + public AmqpInboundGateway(AbstractMessageListenerContainer listenerContainer) { this(listenerContainer, new RabbitTemplate(listenerContainer.getConnectionFactory()), false); } @@ -175,6 +177,16 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { this.recoveryCallback = recoveryCallback; } + /** + * Set to true to bind the source message in the header named + * {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}. + * @param bindSourceMessage true to bind. + * @since 5.1.6 + */ + public void setBindSourceMessage(boolean bindSourceMessage) { + this.bindSourceMessage = bindSourceMessage; + } + @Override public String getComponentType() { return "amqp:inbound-gateway"; @@ -295,6 +307,9 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { if (AmqpInboundGateway.this.retryTemplate != null) { headers.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, new AtomicInteger()); } + if (AmqpInboundGateway.this.bindSourceMessage) { + headers.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, message); + } } catch (RuntimeException e) { MessageChannel errorChannel = getErrorChannel(); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java index dcdcff2845..e62e599753 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java @@ -142,10 +142,11 @@ public class AmqpMessageSource extends AbstractMessageSource { } /** - * Set to true to include the raw spring-amqp message as a header - * with key {@link AmqpMessageHeaderErrorMessageStrategy#AMQP_RAW_MESSAGE}, - * enabling callers to have access to the message to process errors. - * @param rawMessageHeader true to include the header. + * Set to true to include the raw spring-amqp message as a header with key + * {@link AmqpMessageHeaderErrorMessageStrategy#AMQP_RAW_MESSAGE}, enabling callers to + * have access to the message to process errors. The raw message is also added to the + * common header {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}. + * @param rawMessageHeader true to include the headers. */ public void setRawMessageHeader(boolean rawMessageHeader) { this.rawMessageHeader = rawMessageHeader; @@ -180,6 +181,7 @@ public class AmqpMessageSource extends AbstractMessageSource { .setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, callback); if (this.rawMessageHeader) { builder.setHeader(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, amqpMessage); + builder.setHeader(IntegrationMessageHeaderAccessor.SOURCE_DATA, amqpMessage); } return builder; } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceTests.java index 96fc1c415a..901785c6a2 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.amqp.inbound; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.sameInstance; import static org.junit.Assert.assertThat; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.isNull; @@ -32,6 +33,7 @@ import org.junit.Test; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.support.AmqpHeaders; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.acks.AcknowledgmentCallback.Status; import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy; @@ -46,6 +48,7 @@ import com.rabbitmq.client.GetResponse; /** * @author Gary Russell + * @author Artem Bilan * * @since 5.0.1 * @@ -73,6 +76,8 @@ public class AmqpMessageSourceTests { assertThat(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE), instanceOf(org.springframework.amqp.core.Message.class)); assertThat(received.getHeaders().get(AmqpHeaders.CONSUMER_QUEUE), equalTo("foo")); + assertThat(received.getHeaders().get(IntegrationMessageHeaderAccessor.SOURCE_DATA), + sameInstance(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE))); // 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" diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java index 87c2321f95..f1cdc1d4d9 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java @@ -107,6 +107,7 @@ public class InboundEndpointTests { adapter.setOutputChannel(channel); adapter.setBeanFactory(mock(BeanFactory.class)); + adapter.setBindSourceMessage(true); adapter.afterPropertiesSet(); Object payload = new Foo("bar1"); @@ -129,6 +130,8 @@ public class InboundEndpointTests { assertSame(rabbitChannel, result.getHeaders().get(AmqpHeaders.CHANNEL)); assertEquals(123L, result.getHeaders().get(AmqpHeaders.DELIVERY_TAG)); + org.springframework.amqp.core.Message sourceData = StaticMessageHeaderAccessor.getSourceData(result); + assertThat(sourceData).isSameAs(amqpMessage); } @Test @@ -162,6 +165,8 @@ public class InboundEndpointTests { Message result = new JsonToObjectTransformer().transform(receive); assertEquals(payload, result.getPayload()); + org.springframework.amqp.core.Message sourceData = StaticMessageHeaderAccessor.getSourceData(result); + assertThat(sourceData).isNull(); } @Test diff --git a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java index d9f5c40e81..67ed5d48ef 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java @@ -66,6 +66,12 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor { public static final String ACKNOWLEDGMENT_CALLBACK = "acknowledgmentCallback"; + /** + * Raw source message. + */ + public static final String SOURCE_DATA = "sourceData"; + + private Set readOnlyHeaders = new HashSet<>(); public IntegrationMessageHeaderAccessor(@Nullable Message message) { @@ -149,6 +155,18 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor { return getHeader(DELIVERY_ATTEMPT, AtomicInteger.class); } + /** + * Get the source data header, if present. + * @param the data type. + * @return the source header. + * @since 5.1.6 + */ + @SuppressWarnings("unchecked") + @Nullable + public T getSourceData() { + return (T) getHeader(SOURCE_DATA); + } + @SuppressWarnings("unchecked") @Nullable public T getHeader(String key, Class type) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/StaticMessageHeaderAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/StaticMessageHeaderAccessor.java index 717868565b..fa77f6de67 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/StaticMessageHeaderAccessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/StaticMessageHeaderAccessor.java @@ -107,4 +107,10 @@ public final class StaticMessageHeaderAccessor { AcknowledgmentCallback.class); } + @SuppressWarnings("unchecked") + @Nullable + public static T getSourceData(Message message) { + return (T) message.getHeaders().get(IntegrationMessageHeaderAccessor.SOURCE_DATA); + } + }