diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundPolledChannelAdapterSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundPolledChannelAdapterSpec.java index 5cda92ba8f..3d61f32e91 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundPolledChannelAdapterSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundPolledChannelAdapterSpec.java @@ -65,4 +65,9 @@ public class AmqpInboundPolledChannelAdapterSpec return this; } + public AmqpInboundPolledChannelAdapterSpec rawMessageHeader(boolean rawMessageHeader) { + this.target.setRawMessageHeader(rawMessageHeader); + return this; + } + } 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 235df29de1..9cccdf2a92 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 @@ -32,6 +32,7 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.context.MessageSource; import org.springframework.integration.IntegrationMessageHeaderAccessor; 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.endpoint.AbstractMessageSource; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; @@ -66,6 +67,8 @@ public class AmqpMessageSource extends AbstractMessageSource { private MessageConverter messageConverter = new SimpleMessageConverter(); + private boolean rawMessageHeader; + public AmqpMessageSource(ConnectionFactory connectionFactory, String queue) { this(connectionFactory, new AmqpAckCallbackFactory(), queue); } @@ -132,6 +135,20 @@ public class AmqpMessageSource extends AbstractMessageSource { this.messageConverter = messageConverter; } + protected boolean isRawMessageHeader() { + return this.rawMessageHeader; + } + + /** + * 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. + */ + public void setRawMessageHeader(boolean rawMessageHeader) { + this.rawMessageHeader = rawMessageHeader; + } + @Override public String getComponentType() { return "amqp:message-source"; @@ -153,11 +170,15 @@ public class AmqpMessageSource extends AbstractMessageSource { MessageProperties messageProperties = this.propertiesConverter.toMessageProperties(resp.getProps(), resp.getEnvelope(), StandardCharsets.UTF_8.name()); Map headers = this.headerMapper.toHeadersFromRequest(messageProperties); - Object payload = this.messageConverter - .fromMessage(new org.springframework.amqp.core.Message(resp.getBody(), messageProperties)); - return getMessageBuilderFactory().withPayload(payload) + org.springframework.amqp.core.Message amqpMessage = new org.springframework.amqp.core.Message(resp.getBody(), messageProperties); + Object payload = this.messageConverter.fromMessage(amqpMessage); + AbstractIntegrationMessageBuilder builder = getMessageBuilderFactory().withPayload(payload) .copyHeaders(headers) .setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, callback); + if (this.rawMessageHeader) { + builder.setHeader(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, amqpMessage); + } + return builder; } catch (IOException e) { RabbitUtils.closeChannel(channel); 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 0b55e9dab1..8cfba0711c 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 @@ -16,6 +16,8 @@ package org.springframework.integration.amqp.inbound; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertThat; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.BDDMockito.willReturn; @@ -30,6 +32,7 @@ import java.util.concurrent.TimeoutException; import org.junit.Test; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy; import org.springframework.integration.support.AcknowledgmentCallback.Status; import org.springframework.integration.support.StaticMessageHeaderAccessor; import org.springframework.messaging.Message; @@ -65,7 +68,10 @@ public class AmqpMessageSourceTests { CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory); AmqpMessageSource source = new AmqpMessageSource(ccf, "foo"); + source.setRawMessageHeader(true); Message received = source.receive(); + assertThat(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE), + instanceOf(org.springframework.amqp.core.Message.class)); // 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/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 66f8e10571..e9413dfedf 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -301,6 +301,38 @@ public class AmqpJavaApplication { } ---- +=== Polled Inbound Channel Adapter + +Starting with _version 5.0.1_, a polled channel adapter is provided, allowing fetching individual messages on-demand, for example with a `MessageSourcePollingTemplate` or a poller. +See <> for more information. + +It does not currently have XML configuration. + +[source, java] +---- +@Bean +public AmqpMessageSource source(ConnectionFactory connectionFactory) { + return new AmpqpMessageSource(connectionFactory, "someQueue"); +} +---- + +Refer to the javadocs for configuration properties. + +With the Java DSL: + +[source, java] +---- +@Bean +public IntegrationFlow flow() { + return IntegrationFlows.from(Amqp.inboundPolledAdapter(connectionFactory(), DSL_QUEUE), + e -> e.poller(Pollers.fixedDelay(1_000)).autoStartup(false)) + .handle(p -> { + ... + }) + .get(); +} +---- + [[amqp-inbound-gateway]] === Inbound Gateway