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

AMQP Docs: Add JavaConfig and DSL samples

Browse Source
This commit is contained in:
Gary Russell
2015-10-01 14:12:53 -04:00
committed by Artem Bilan
parent d09431a951
commit b2d04bf7de
1 changed files with 415 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

426
src/reference/asciidoc/amqp.adoc
View File

@@ -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 `<bean/>` d
<bean id="container"
class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="queueNames" value="foo.queue" />
<property name="defaultRequeueRejected" value="false"/>
<property name="connectionFactory" ref="connectionFactory" />
<property name="queueNames" value="foo.queue" />
<property name="defaultRequeueRejected" value="false"/>
</bean>
----
=====
@@ -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 <<amqp-inbound-channel-adapter>> 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