diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 45dc99d0bd..0b16329dba 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -145,25 +145,27 @@ _Optional_. <17> Tells the AMQP broker how many messages to send to each consumer in a single request. Often this can be set quite high to improve throughput. -It should be greater than or equal to the transaction size (see attribute "tx-size")._Optional (Defaults to 1)_. +It should be greater than or equal to the transaction size (see attribute "tx-size"). +_Optional (Defaults to 1)_. <18> Receive timeout in milliseconds. _Optional (Defaults to 1000)_. -<19> Specifies the interval between recovery attempts of the underlying `SimpleMessageListenerContainer` (in milliseconds)._Optional (Defaults to 5000)_. +<19> Specifies the interval between recovery attempts of the underlying `SimpleMessageListenerContainer` (in milliseconds) +._Optional (Defaults to 5000)_. <20> If 'true', and none of the queues are available on the broker, the container will throw a fatal exception during startup and will stop if the queues are deleted when the container is running (after making 3 attempts to passively declare the queues). -If false, the container will not throw an exception and go into recovery mode, attempting to restart according to the `revcovery-interval`. +If false, the container will not throw an exception and go into recovery mode, attempting to restart according to the `recovery-interval`. _Optional (Defaults to `true`)_. <21> The time to wait for workers in milliseconds after the underlying `SimpleMessageListenerContainer` is stopped, and before the AMQP connection is forced closed. If any workers are active when the shutdown signal comes they will be allowed to finish processing as long as they can finish within this timeout. Otherwise the connection is closed and messages remain unacked (if the channel is transactional). -Defaults to 5000 milliseconds._Optional (Defaults to 5000)_. +_Optional (Defaults to 5000)_. <22> By default, the underlying `SimpleMessageListenerContainer` uses a SimpleAsyncTaskExecutor implementation, that fires up a new Thread for each task, executing it asynchronously. @@ -181,7 +183,9 @@ _Optional (Defaults to DefaultTransactionAttribute)_. The transaction manager works in conjunction with the "channel-transacted" attribute. If there is already a transaction in progress when the framework is sending or receiving a message, and the channelTransacted flag is true, then the commit or rollback of the messaging transaction will be deferred until the end of the current transaction. If the channelTransacted flag is false, then no transaction semantics apply to the messaging operation (it is auto-acked). -For further information see chapter 1.9 of the Spring AMQP reference guide: http://static.springsource.org/spring-amqp/docs/1.0.x/reference/html/#d0e525 _Optional_. +For further information see +http://docs.spring.io/spring-amqp/reference/html/%5Freference.html#%5Ftransactions[Transactions with Spring AMQP]. +_Optional_. <25> Tells the `SimpleMessageListenerContainer` how many messages to process in a single transaction (if the channel is transactional). @@ -201,9 +205,9 @@ For this reason, you must define the container using a normal Spring `` d - - - + + + ---- ===== @@ -213,6 +217,84 @@ The JMS Inbound Channel Adapter is using a JmsDestinationPollingSource under the The AMQP Inbound Channel Adapter on the other side uses a`SimpleMessageListenerContainer` and is message driven. In that regard it is more similar to the JMS Message Driven Channel Adapter. +==== Configuring with JavaConfig + +The following Spring Boot application provides an example of configuring the inbound adapter using Java configuration: +[source, java] +---- +@SpringBootApplication +public class AmqpJavaApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder(AmqpJavaApplication.class) + .web(false) + .run(args); + } + + @Bean + public MessageChannel amqpInputChannel() { + return new DirectChannel(); + } + + @Bean + public AmqpInboundChannelAdapter inbound(SimpleMessageListenerContainer listenerContainer, + @Qualifier("amqpInputChannel") MessageChannel channel) { + AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer); + adapter.setOutputChannel(channel); + return adapter; + } + + @Bean + public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory) { + SimpleMessageListenerContainer container = + new SimpleMessageListenerContainer(connectionFactory); + container.setQueueNames("foo"); + container.setConcurrentConsumers(2); + // ... + return container; + } + + @Bean + @ServiceActivator(inputChannel = "amqpInputChannel") + public MessageHandler handler() { + return new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + System.out.println(message.getPayload()); + } + + }; + } + +} +---- + +==== Configuring with the Java DSL + +The following Spring Boot application provides an example of configuring the inbound adapter using the Java DSL: + +[source, java] +---- +@SpringBootApplication +public class AmqpJavaApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder(AmqpJavaApplication.class) + .web(false) + .run(args); + } + + @Bean + public IntegrationFlow amqpInbound(ConnectionFactory connectionFactory) { + return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "foo")) + .handle(m -> System.out.println(m.getPayload())) + .get(); + } + +} +---- + [[amqp-inbound-gateway]] === Inbound Gateway @@ -283,6 +365,85 @@ if you anticipate cases when no `replyTo` property exists in the request message See the note in <> about configuring the `listener-container` attribute. +==== Configuring with JavaConfig + +The following Spring Boot application provides an example of configuring the inbound gateway using Java configuration: +[source, java] +---- +@SpringBootApplication +public class AmqpJavaApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder(AmqpJavaApplication.class) + .web(false) + .run(args); + } + + @Bean + public MessageChannel amqpInputChannel() { + return new DirectChannel(); + } + + @Bean + public AmqpInboundGateway inbound(SimpleMessageListenerContainer listenerContainer, + @Qualifier("amqpInputChannel") MessageChannel channel) { + AmqpInboundGateway gateway = new AmqpInboundGateway(listenerContainer); + gateway.setRequestChannel(channel); + gateway.setDefaultReplyTo("bar"); + return gateway; + } + + @Bean + public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory) { + SimpleMessageListenerContainer container = + new SimpleMessageListenerContainer(connectionFactory); + container.setQueueNames("foo"); + container.setConcurrentConsumers(2); + // ... + return container; + } + + @Bean + @ServiceActivator(inputChannel = "amqpInputChannel") + public MessageHandler handler() { + return new AbstractReplyProducingMessageHandler() { + + @Override + protected Object handleRequestMessage(Message requestMessage) { + return "reply to " + requestMessage.getPayload(); + } + + }; + } + +} +---- + +==== Configuring with the Java DSL + +The following Spring Boot application provides an example of configuring the inbound gateway using the Java DSL: + +[source, java] +---- +@SpringBootApplication +public class AmqpJavaApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder(AmqpJavaApplication.class) + .web(false) + .run(args); + } + + @Bean // return the upper cased payload + public IntegrationFlow amqpInboundGateway(ConnectionFactory connectionFactory) { + return IntegrationFlows.from(Amqp.inboundGateway(connectionFactory, "foo")) + .transform(String.class, String::toUpperCase) + .get(); + } + +} +---- + [[amqp-inbound-ack]] === Inbound Endpoint Acknowledge Mode @@ -447,6 +608,87 @@ Using a `return-channel` requires a `RabbitTemplate` with the `mandatory` proper When using multiple outbound endpoints with returns, a separate `RabbitTemplate` is needed for each endpoint. ===== +==== Configuring with JavaConfig + +The following Spring Boot application provides an example of configuring the outbound adapter using Java configuration: +[source, java] +---- +@SpringBootApplication +@IntegrationComponentScan +public class AmqpJavaApplication { + + public static void main(String[] args) { + ConfigurableApplicationContext context = + new SpringApplicationBuilder(AmqpJavaApplication.class) + .web(false) + .run(args); + MyGateway gateway = context.getBean(MyGateway.class); + gateway.sendToRabbit("foo"); + } + + @Bean + @ServiceActivator(inputChannel = "amqpOutboundChannel") + public AmqpOutboundEndpoint amqpOutbound(AmqpTemplate amqpTemplate) { + AmqpOutboundEndpoint outbound = new AmqpOutboundEndpoint(amqpTemplate); + outbound.setRoutingKey("foo"); // default exchange - route to queue 'foo' + return outbound; + } + + @Bean + public MessageChannel amqpOutboundChannel() { + return new DirectChannel(); + } + + @MessagingGateway(defaultRequestChannel = "amqpOutboundChannel") + public interface MyGateway { + + void sendToRabbit(String data); + + } + +} +---- + +==== Configuring with the Java DSL + +The following Spring Boot application provides an example of configuring the outbound adapter using the Java DSL: + +[source, java] +---- +@SpringBootApplication +@IntegrationComponentScan +public class AmqpJavaApplication { + + public static void main(String[] args) { + ConfigurableApplicationContext context = + new SpringApplicationBuilder(AmqpJavaApplication.class) + .web(false) + .run(args); + MyGateway gateway = context.getBean(MyGateway.class); + gateway.sendToRabbit("foo"); + } + + @Bean + public IntegrationFlow amqpOutbound(AmqpTemplate amqpTemplate) { + return IntegrationFlows.from(amqpOutboundChannel()) + .handle(Amqp.outboundAdapter(amqpTemplate) + .routingKey("foo")) // default exchange - route to queue 'foo' + .get(); + } + + @Bean + public MessageChannel amqpOutboundChannel() { + return new DirectChannel(); + } + + @MessagingGateway(defaultRequestChannel = "amqpOutboundChannel") + public interface MyGateway { + + void sendToRabbit(String data); + + } +---- + [[amqp-outbound-gateway]] === Outbound Gateway @@ -515,7 +757,8 @@ Default: `true`. <10> The routing-key to use when sending Messages. By default, this will be an empty String. -Mutually exclusive with 'routing-key-expression'_Optional_. +Mutually exclusive with 'routing-key-expression'. +_Optional_. <11> A SpEL expression that is evaluated to determine the routing-key to use when sending Messages, with the message as the root object (e.g. @@ -529,7 +772,8 @@ _Optional_. Overridden if the 'header-mapper' sets the delivery mode. The 'DefaultHeaderMapper' sets the value if the Spring Integration message header `amqp_deliveryMode` is present. If this attribute is not supplied and the header mapper doesn't set it, the default depends on the underlying spring-amqp 'MessagePropertiesConverter' used by the 'RabbitTemplate'. -If that is not customized at all, the default is 'PERSISTENT'._Optional_. +If that is not customized at all, the default is 'PERSISTENT'. +_Optional_. <13> Since _version 4.2_. An expression defining correlation data. When provided, this configures the underlying amqp template to receive publisher confirms. @@ -572,6 +816,95 @@ When using multiple outbound endpoints with returns, a separate `RabbitTemplate` IMPORTANT: The underlying `AmqpTemplate` has a default `replyTimeout` of 5 seconds. If you require a longer timeout, it must be configured on the `template`. +==== Configuring with JavaConfig + +The following Spring Boot application provides an example of configuring the outbound gateway using Java configuration: +[source, java] +---- +@SpringBootApplication +@IntegrationComponentScan +public class AmqpJavaApplication { + + public static void main(String[] args) { + ConfigurableApplicationContext context = + new SpringApplicationBuilder(AmqpJavaApplication.class) + .web(false) + .run(args); + MyGateway gateway = context.getBean(MyGateway.class); + String reply = gateway.sendToRabbit("foo"); + System.out.println(reply); + } + + @Bean + @ServiceActivator(inputChannel = "amqpOutboundChannel") + public AmqpOutboundEndpoint amqpOutbound(AmqpTemplate amqpTemplate) { + AmqpOutboundEndpoint outbound = new AmqpOutboundEndpoint(amqpTemplate); + outbound.setExpectReply(true); + outbound.setRoutingKey("foo"); // default exchange - route to queue 'foo' + return outbound; + } + + @Bean + public MessageChannel amqpOutboundChannel() { + return new DirectChannel(); + } + + @MessagingGateway(defaultRequestChannel = "amqpOutboundChannel") + public interface MyGateway { + + String sendToRabbit(String data); + + } + +} +---- + +Notice that the only difference between the outbound adapter and outbound gateway configuration is the setting of the +`expectReply` property. + +==== Configuring with the Java DSL + +The following Spring Boot application provides an example of configuring the outbound adapter using the Java DSL: + +[source, java] +---- +@SpringBootApplication +@IntegrationComponentScan +public class AmqpJavaApplication { + + public static void main(String[] args) { + ConfigurableApplicationContext context = + new SpringApplicationBuilder(AmqpJavaApplication.class) + .web(false) + .run(args); + RabbitTemplate template = context.getBean(RabbitTemplate.class); + MyGateway gateway = context.getBean(MyGateway.class); + String reply = gateway.sendToRabbit("foo"); + System.out.println(reply); + } + + @Bean + public IntegrationFlow amqpOutbound(AmqpTemplate amqpTemplate) { + return IntegrationFlows.from(amqpOutboundChannel()) + .handle(Amqp.outboundGateway(amqpTemplate) + .routingKey("foo")) // default exchange - route to queue 'foo' + .get(); + } + + @Bean + public MessageChannel amqpOutboundChannel() { + return new DirectChannel(); + } + + @MessagingGateway(defaultRequestChannel = "amqpOutboundChannel") + public interface MyGateway { + + String sendToRabbit(String data); + + } +} +---- + [[amqp-channels]] === AMQP Backed Message Channels @@ -602,12 +935,83 @@ A publish/subscribe channel would look like this: ---- Under the covers a Fanout Exchange named "si.fanout.pubSubChannel" would be declared, and this channel will send to that Fanout Exchange. -This channel will also declare a server-named exclusive, autodelete, non-durable Queue and bind that to the Fanout Exchange while registering a consumer on that Queue to receive Messages. +This channel will also declare a server-named exclusive, auto-delete, non-durable Queue and bind that to the Fanout Exchange while registering a consumer on that Queue to receive Messages. There is no "pollable" option for a publish-subscribe-channel; it must be message-driven. Starting with _version 4.1_ AMQP Backed Message Channels, alongside with `channel-transacted`, support `template-channel-transacted` to separate `transactional` configuration for the `AbstractMessageListenerContainer` and for the `RabbitTemplate`. Note, previously, the `channel-transacted` was `true` by default, now it changed to `false` as standard default value for the `AbstractMessageListenerContainer`. +==== Configuring with JavaConfig + +The following provides an example of configuring the channels using Java configuration: + +[source, java] +---- +@Bean +public AmqpChannelFactoryBean pollable(ConnectionFactory connectionFactory) { + AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(); + factoryBean.setConnectionFactory(connectionFactory); + factoryBean.setQueueName("foo"); + factoryBean.setPubSub(false); + return factoryBean; +} + +@Bean +public AmqpChannelFactoryBean messageDriven(ConnectionFactory connectionFactory) { + AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(true); + factoryBean.setConnectionFactory(connectionFactory); + factoryBean.setQueueName("bar"); + factoryBean.setPubSub(false); + return factoryBean; +} + +@Bean +public AmqpChannelFactoryBean pubSub(ConnectionFactory connectionFactory) { + AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(true); + factoryBean.setConnectionFactory(connectionFactory); + factoryBean.setQueueName("baz"); + factoryBean.setPubSub(false); + return factoryBean; +} +---- + +==== Configuring with the Java DSL + +The following provides an example of configuring the channels using the Java DSL: + +[source, java] +---- +@Bean +public IntegrationFlow pollableInFlow(ConnectionFactory connectionFactory) { + return IntegrationFlows.from(...) + ... + .channel(Amqp.pollableChannel(connectionFactory) + .queueName("foo")) + ... + .get(); +} + +@Bean +public IntegrationFlow messageDrivenInFow(ConnectionFactory connectionFactory) { + return IntegrationFlows.from(...) + ... + .channel(Amqp.channel(connectionFactory) + .queueName("bar")) + ... + .get(); +} + +@Bean +public IntegrationFlow pubSubInFlow(ConnectionFactory connectionFactory) { + return IntegrationFlows.from(...) + ... + .channel(Amqp.publisSubscribeChannel(connectionFactory) + .queueName("baz")) + ... + .get(); +} +---- + [[amqp-message-headers]] === AMQP Message Headers