.transform(String.class, String::toUpperCase)
.get();
}
@Bean
public IntegrationFlow jmsOutboundGatewayFlow() {
return IntegrationFlows.from("jmsOutboundGatewayChannel")
.handle(Jms.outboundGateway(this.jmsConnectionFactory)
.replyContainer(c ->
c.concurrentConsumers(3)
.sessionTransacted(true))
.requestDestination("jmsPipelineTest"))
.get();
}
@Bean
public IntegrationFlow sendMailFlow() {
return IntegrationFlows.from("sendMailChannel")
.handle(Mail.outboundAdapter("localhost")
.port(smtpPort)
.credentials("user", "pw")
.protocol("smtp")
.javaMailProperties(p -> p.put("mail.debug", "true")),
e -> e.id("sendMailEndpoint"))
.get();
}
The preceding example shows how to use the “namespace factories” as inline adapters declarations.
However, you can use them from @Bean
definitions to make the IntegrationFlow
method chain more readable.
| We are soliciting community feedback on these namespace factories before we spend effort on others. We also appreciate any input into prioritization for which adapters and gateways we should support next. |
You can find more Java DSL samples in the protocol-specific chapters throughout this reference manual.
All other protocol channel adapters may be configured as generic beans and wired to the IntegrationFlow
, as the following examples show:
@Bean
public QueueChannelSpec wrongMessagesChannel() {
return MessageChannels
.queue()
.wireTap("wrongMessagesWireTapChannel");
}
@Bean
public IntegrationFlow xpathFlow(MessageChannel wrongMessagesChannel) {
return IntegrationFlows.from("inputChannel")
.filter(new StringValueTestXPathMessageSelector("namespace-uri(/*)", "my:namespace"),
e -> e.discardChannel(wrongMessagesChannel))
.log(LoggingHandler.Level.ERROR, "test.category", m -> m.getHeaders().getId())
.route(xpathRouter(wrongMessagesChannel))
.get();
}
@Bean
public AbstractMappingMessageRouter xpathRouter(MessageChannel wrongMessagesChannel) {
XPathRouter router = new XPathRouter("local-name(/*)");
router.setEvaluateAsString(true);
router.setResolutionRequired(false);
router.setDefaultOutputChannel(wrongMessagesChannel);
router.setChannelMapping("Tags", "splittingChannel");
router.setChannelMapping("Tag", "receivedChannel");
return router;
}
11.17. IntegrationFlowAdapter
The IntegrationFlow
interface can be implemented directly and specified as a component for scanning, as the following example shows:
@Component
public class MyFlow implements IntegrationFlow {
@Override
public void configure(IntegrationFlowDefinition<?> f) {
f.<String, String>transform(String::toUpperCase);
}
}
It is picked up by the IntegrationFlowBeanPostProcessor
and correctly parsed and registered in the application context.
For convenience and to gain the benefits of loosely coupled architecture, we provide the IntegrationFlowAdapter
base class implementation.
It requires a buildFlow()
method implementation to produce an IntegrationFlowDefinition
by using one of from()
methods, as the following example shows:
@Component
public class MyFlowAdapter extends IntegrationFlowAdapter {
private final AtomicBoolean invoked = new AtomicBoolean();
public Date nextExecutionTime(TriggerContext triggerContext) {
return this.invoked.getAndSet(true) ? null : new Date();
}
@Override
protected IntegrationFlowDefinition<?> buildFlow() {
return from(this, "messageSource",
e -> e.poller(p -> p.trigger(this::nextExecutionTime)))
.split(this)
.transform(this)
.aggregate(a -> a.processor(this, null), null)
.enrichHeaders(Collections.singletonMap("thing1", "THING1"))
.filter(this)
.handle(this)
.channel(c -> c.queue("myFlowAdapterOutput"));
}
public String messageSource() {
return "T,H,I,N,G,2";
}
@Splitter
public String[] split(String payload) {
return StringUtils.commaDelimitedListToStringArray(payload);
}
@Transformer
public String transform(String payload) {
return payload.toLowerCase();
}
@Aggregator
public String aggregate(List<String> payloads) {
return payloads.stream().collect(Collectors.joining());
}
@Filter
public boolean filter(@Header Optional<String> thing1) {
return thing1.isPresent();
}
@ServiceActivator
public String handle(String payload, @Header String thing1) {
return payload + ":" + thing1;
}
}
11.18. Dynamic and Runtime Integration Flows
IntegrationFlow
and all its dependent components can be registered at runtime.
Before version 5.0, we used the BeanFactory.registerSingleton()
hook.
Starting in the Spring Framework 5.0
, we use the instanceSupplier
hook for programmatic BeanDefinition
registration.
The following example shows how to programmatically register a bean:
BeanDefinition beanDefinition =
BeanDefinitionBuilder.genericBeanDefinition((Class<Object>) bean.getClass(), () -> bean)
.getRawBeanDefinition();
((BeanDefinitionRegistry) this.beanFactory).registerBeanDefinition(beanName, beanDefinition);
Note that, in the preceding example, the instanceSupplier
hook is the last parameter to the genericBeanDefinition
method, provided by a lambda in this case.
All the necessary bean initialization and lifecycle is done automatically, as it is with the standard context configuration bean definitions.
To simplify the development experience, Spring Integration introduced IntegrationFlowContext
to register and manage IntegrationFlow
instances at runtime, as the following example shows:
@Autowired
private AbstractServerConnectionFactory server1;
@Autowired
private IntegrationFlowContext flowContext;
...
@Test
public void testTcpGateways() {
TestingUtilities.waitListening(this.server1, null);
IntegrationFlow flow = f -> f
.handle(Tcp.outboundGateway(Tcp.netClient("localhost", this.server1.getPort())
.serializer(TcpCodecs.crlf())
.deserializer(TcpCodecs.lengthHeader1())
.id("client1"))
.remoteTimeout(m -> 5000))
.transform(Transformers.objectToString());
IntegrationFlowRegistration theFlow = this.flowContext.registration(flow).register();
assertThat(theFlow.getMessagingTemplate().convertSendAndReceive("foo", String.class), equalTo("FOO"));
}
This is useful when we have multiple configuration options and have to create several instances of similar flows.
To do so, we can iterate our options and create and register IntegrationFlow
instances within a loop.
Another variant is when our source of data is not Spring-based and we must create it on the fly.
Such a sample is Reactive Streams event source, as the following example shows:
Flux<Message<?>> messageFlux =
Flux.just("1,2,3,4")
.map(v -> v.split(","))
.flatMapIterable(Arrays::asList)
.map(Integer::parseInt)
.map(GenericMessage<Integer>::new);
QueueChannel resultChannel = new QueueChannel();
IntegrationFlow integrationFlow =
IntegrationFlows.from(messageFlux)
.<Integer, Integer>transform(p -> p * 2)
.channel(resultChannel)
.get();
this.integrationFlowContext.registration(integrationFlow)
.register();
The IntegrationFlowRegistrationBuilder
(as a result of the IntegrationFlowContext.registration()
) can be used to specify a bean name for the IntegrationFlow
to register, to control its autoStartup
, and to register, non-Spring Integration beans.
Usually, those additional beans are connection factories (AMQP, JMS, (S)FTP, TCP/UDP, and others.), serializers and deserializers, or any other required support components.
You can use the IntegrationFlowRegistration.destroy()
callback to remove a dynamically registered IntegrationFlow
and all its dependent beans when you no longer need them.
See the IntegrationFlowContext
Javadoc for more information.
Starting with version 5.0.6, all generated bean names in an IntegrationFlow definition are prepended with the flow ID as a prefix.
We recommend always specifying an explicit flow ID.
Otherwise, a synchronization barrier is initiated in the IntegrationFlowContext , to generate the bean name for the IntegrationFlow and register its beans.
We synchronize on these two operations to avoid a race condition when the same generated bean name may be used for different IntegrationFlow instances.
|
Also, starting with version 5.0.6, the registration builder API has a new method: useFlowIdAsPrefix()
.
This is useful if you wish to declare multiple instances of the same flow and avoid bean name collisions when components in the flows have the same ID, as the following example shows:
private void registerFlows() {
IntegrationFlowRegistration flow1 =
this.flowContext.registration(buildFlow(1234))
.id("tcp1")
.useFlowIdAsPrefix()
.register();
IntegrationFlowRegistration flow2 =
this.flowContext.registration(buildFlow(1235))
.id("tcp2")
.useFlowIdAsPrefix()
.register();
}
private IntegrationFlow buildFlow(int port) {
return f -> f
.handle(Tcp.outboundGateway(Tcp.netClient("localhost", port)
.serializer(TcpCodecs.crlf())
.deserializer(TcpCodecs.lengthHeader1())
.id("client"))
.remoteTimeout(m -> 5000))
.transform(Transformers.objectToString());
}
In this case, the message handler for the first flow can be referenced with bean a name of tcp1.client.handler
.
An id attribute is required when you usE useFlowIdAsPrefix() .
|
11.19. IntegrationFlow
as Gateway
The IntegrationFlow
can start from the service interface that provides a GatewayProxyFactoryBean
component, as the following example shows:
public interface ControlBusGateway {
void send(String command);
}
...
@Bean
public IntegrationFlow controlBusFlow() {
return IntegrationFlows.from(ControlBusGateway.class)
.controlBus()
.get();
}
All the proxy for interface methods are supplied with the channel to send messages to the next integration component in the IntegrationFlow
.
You can mark the service interface with the @MessagingGateway
annotation and mark the methods with the @Gateway
annotations.
Nevertheless, the requestChannel
is ignored and overridden with that internal channel for the next component in the IntegrationFlow
.
Otherwise, creating such a configuration by using IntegrationFlow
does not make sense.
By default a GatewayProxyFactoryBean
gets a conventional bean name, such as [FLOW_BEAN_NAME.gateway]
.
You can change that ID by using the @MessagingGateway.name()
attribute or the overloaded from(Class<?> serviceInterface, String beanName)
factory method.
With Java 8, you can even create an integration fateway with the java.util.function
interfaces, as the following example shows:
@Bean
public IntegrationFlow errorRecovererFlow() {
return IntegrationFlows.from(Function.class, "errorRecovererFunction")
.handle((GenericHandler<?>) (p, h) -> {
throw new RuntimeException("intentional");
}, e -> e.advice(retryAdvice()))
.get();
}
That errorRecovererFlow
can be used as follows:
@Autowired
@Qualifier("errorRecovererFunction")
private Function<String, String> errorRecovererFlowGateway;
12. System Management
12.1. Metrics and Management
This section describes how to capture metrics for Spring Integration. In recent versions, we have relied more on Micrometer (see micrometer.io), and we plan to use Micrometer even more in future releases.
12.1.1. Configuring Metrics Capture
| Prior to version 4.2, metrics were only available when JMX was enabled. See JMX Support. |
To enable MessageSource
, MessageChannel
, and MessageHandler
metrics, add an <int:management/>
bean to the application context (in XML) or annotate one of your @Configuration
classes with @EnableIntegrationManagement
(in Java).
MessageSource
instances maintain only counts, MessageChannel
instances and MessageHandler
instances maintain duration statistics in addition to counts.
See MessageChannel
Metric Features and MessageHandler Metric Features, later in this chapter.
Doing so causes the automatic registration of the IntegrationManagementConfigurer
bean in the application context.
Only one such bean can exist in the context, and, if registered manually via a <bean/>
definition, it must have the bean name set to integrationManagementConfigurer
.
This bean applies its configuration to beans after all beans in the context have been instantiated.
In addition to metrics, you can control debug logging in the main message flow.
In very high volume applications, even calls to isDebugEnabled()
can be quite expensive with some logging subsystems.
You can disable all such logging to avoid this overhead.
Exception logging (debug or otherwise) is not affected by this setting.
The following listing shows the available options for controlling logging:
<int:management
default-logging-enabled="true" (1)
default-counts-enabled="false" (2)
default-stats-enabled="false" (3)
counts-enabled-patterns="foo, !baz, ba*" (4)
stats-enabled-patterns="fiz, buz" (5)
metrics-factory="myMetricsFactory" /> (6)
@Configuration
@EnableIntegration
@EnableIntegrationManagement(
defaultLoggingEnabled = "true", (1)
defaultCountsEnabled = "false", (2)
defaultStatsEnabled = "false", (3)
countsEnabled = { "foo", "${count.patterns}" }, (4)
statsEnabled = { "qux", "!*" }, (5)
MetricsFactory = "myMetricsFactory") (6)
public static class ContextConfiguration {
...
}
| 1 | Set to false to disable all logging in the main message flow, regardless of the log system category settings.
Set to 'true' to enable debug logging (if also enabled by the logging subsystem).
Only applied if you have not explicitly configured the setting in a bean definition.
The default is true . |
| 2 | Enable or disable count metrics for components that do not match one of the patterns in <4>.
Only applied if you have not explicitly configured the setting in a bean definition.
The default is false . |
| 3 | Enable or disable statistical metrics for components that do not match one of the patterns in <5>. Only applied if you have not explicitly configured the setting in a bean definition. The default is 'false'. |
| 4 | A comma-delimited list of patterns for beans for which counts should be enabled.
You can negate the pattern with ! .
First match (positive or negative) wins.
In the unlikely event that you have a bean name starting with ! , escape the ! in the pattern.
For example, \!something positively matches a bean named !something . |
| 5 | A comma-delimited list of patterns for beans for which statistical metrics should be enabled.
You can negate the pattern\ with ! .
First match (positive or negative) wins.
In the unlikely event that you have a bean name starting with ! , escape the ! in the pattern.
\!something positively matches a bean named !something .
The collection of statistics implies the collection of counts. |
| 6 | A reference to a MetricsFactory .
See Metrics Factory. |
At runtime, counts and statistics can be obtained by calling getChannelMetrics
, getHandlerMetrics
and getSourceMetrics
(all from the IntegrationManagementConfigurer
class), which return MessageChannelMetrics
, MessageHandlerMetrics
, and MessageSourceMetrics
, respectively.
See the Javadoc for complete information about these classes.
When JMX is enabled (see JMX Support), IntegrationMBeanExporter
also exposes these metrics.
IMPORTANT:
defaultLoggingEnabled
, defaultCountsEnabled
, and defaultStatsEnabled
are applied only if you have not explicitly configured the corresponding setting in a bean definition.
Starting with version 5.0.2, the framework automatically detects whether the application context has a single MetricsFactory
bean and, if so, uses it instead of the default metrics factory.
12.1.2. Micrometer Integration
Starting with version 5.0.3, the presence of a Micrometer MeterRegistry
in the application context triggers support for Micrometer metrics in addition to the built-in metrics (note that built-in metrics will be removed in a future release).
Micrometer was first supported in version 5.0.2, but changes were made to the Micrometer Meters in version 5.0.3 to make them more suitable for use in dimensional systems.
Further changes were made in 5.0.4.
If you use Micrometer, a minimum of version 5.0.4 is recommended, since some of the changes in 5.0.4 were breaking API changes.
|
To use Micrometer, add one of the MeterRegistry
beans to the application context.
If the IntegrationManagementConfigurer
detects exactly one MeterRegistry
bean, it configures a MicrometerMetricsCaptor
bean with a name of integrationMicrometerMetricsCaptor
.
For each MessageHandler
and MessageChannel
, timers are registered.
For each MessageSource
, a counter is registered.
This only applies to objects that extend AbstractMessageHandler
, AbstractMessageChannel
, and AbstractMessageSource
(which is the case for most framework components).
With Micrometer metrics, the statsEnabled
flag has no effect, since statistics capture is delegated to Micrometer.
The countsEnabled
flag controls whether the Micrometer Meter
instances are updated when processing each message.
The Timer
Meters for send operations on message channels have the following names or tags:
-
name
:spring.integration.send
-
tag
:type:channel
-
tag
:name:<componentName>
-
tag
:result:(success|failure)
-
tag
:exception:(none|exception simple class name)
-
description
:Send processing time
(A failure
result with a none
exception means the channel’s send()
operation returned false
.)
The Counter
Meters for receive operations on pollable message channels have the following names or tags:
-
name
:spring.integration.receive
-
tag
:type:channel
-
tag
:name:<componentName>
-
tag
:result:(success|failure)
-
tag
:exception:(none|exception simple class name)
-
description
:Messages received
The Timer
Meters for operations on message handlers have the following names or tags:
-
name
:spring.integration.send
-
tag
:type:handler
-
tag
:name:<componentName>
-
tag
:result:(success|failure)
-
tag
:exception:(none|exception simple class name)
-
description
:Send processing time
The Counter
meters for message sources have the following names/tags:
-
name
:spring.integration.receive
-
tag
:type:source
-
tag
:name:<componentName>
-
tag
:result:success
-
tag
:exception:none
-
description
:Messages received
In addition, there are three Gauge
Meters:
-
spring.integration.channels
: The number ofMessageChannels
in the application. -
spring.integration.handlers
: The number ofMessageHandlers
in the application. -
spring.integration.sources
: The number ofMessageSources
in the application.
It is possible to customize the names and tags of Meters
created by integration components by providing a subclass of MicrometerMetricsCaptor
.
The MicrometerCustomMetricsTests test case shows a simple example of how to do that.
You can also further customize the meters by overloading the build()
methods on builder subclasses.
12.1.3. MessageChannel
Metric Features
These legacy metrics will be removed in a future release. See Micrometer Integration.
Message channels report metrics according to their concrete type.
If you are looking at a DirectChannel
, you see statistics for the send operation.
If it is a QueueChannel
, you also see statistics for the receive operation as well as the count of messages that are currently buffered by this QueueChannel
.
In both cases, some metrics are simple counters (message count and error count), and some are estimates of averages of interesting quantities.
The algorithms used to calculate these estimates are described briefly in the following table.
| Metric Type | Example | Algorithm |
|---|---|---|
Count |
Send Count |
Simple incrementer. Increases by one when an event occurs. |
Error Count |
Send Error Count |
Simple incrementer. Increases by one when an send results in an error. |
Duration |
Send Duration (method execution time in milliseconds) |
Exponential moving average with decay factor (ten by default). Average of the method execution time over roughly the last ten (by default) measurements. |
Rate |
Send Rate (number of operations per second) |
Inverse of Exponential moving average of the interval between events with decay in time (lapsing over 60 seconds by default) and per measurement (last ten events by default). |
Error Rate |
Send Error Rate (number of errors per second) |
Inverse of exponential moving average of the interval between error events with decay in time (lapsing over 60 seconds by default) and per measurement (last ten events by default). |
Ratio |
Send Success Ratio (ratio of successful to total sends) |
Estimate the success ratio as the exponential moving average of the series composed of values (1 for success and 0 for failure, decaying as per the rate measurement over time and events by default). The error ratio is: 1 - success ratio. |
12.1.4. MessageHandler Metric Features
These legacy metrics will be removed in a future release. See Micrometer Integration.
The following table shows the statistics maintained for message handlers. Some metrics are simple counters (message count and error count), and one is an estimate of averages of send duration. The algorithms used to calculate these estimates are described briefly in the following table:
| Metric Type | Example | Algorithm |
|---|---|---|
Count |
Handle Count |
Simple incrementer. Increases by one when an event occurs. |
Error Count |
Handler Error Count |
Simple incrementer. Increases by one when an invocation results in an error. |
Active Count |
Handler Active Count |
Indicates the number of currently active threads currently invoking the handler (or any downstream synchronous flow). |
Duration |
Handle Duration (method execution time in milliseconds) |
Exponential moving average with decay factor (ten by default). Average of the method execution time over roughly the last ten (default) measurements. |
12.1.5. Time-Based Average Estimates
A feature of the time-based average estimates is that they decay with time if no new measurements arrive. To help interpret the behavior over time, the time (in seconds) since the last measurement is also exposed as a metric.
There are two basic exponential models: decay per measurement (appropriate for duration and anything where the number of measurements is part of the metric) and decay per time unit (more suitable for rate measurements where the time in between measurements is part of the metric).
Both models depend on the fact that S(n) = sum(i=0,i=n) w(i) x(i)
has a special form when w(i) = r^i
, with r=constant
: S(n) = x(n) + r S(n-1)
(so you only have to store S(n-1)
(not the whole series x(i)
) to generate a new metric estimate from the last measurement).
The algorithms used in the duration metrics use r=exp(-1/M)
with M=10
.
The net effect is that the estimate, S(n)
, is more heavily weighted to recent measurements and is composed roughly of the last M
measurements.
So M
is the “window” or lapse rate of the estimate.
For the vanilla moving average, i
is a counter over the number of measurements.
For the rate, we interpret i
as the elapsed time or a combination of elapsed time and a counter (so the metric estimate contains contributions roughly from the last M
measurements and the last T
seconds).
12.1.6. Metrics Factory
A strategy interface MetricsFactory
has been introduced to let you provide custom channel metrics for your MessageChannel
instances and MessageHandler
instances.
By default, a DefaultMetricsFactory
provides a default implementation of MessageChannelMetrics
and MessageHandlerMetrics
, described earlier.
To override the default MetricsFactory
, configure it as described earlier, by providing a reference to your MetricsFactory
bean instance.
You can either customize the default implementations, as described in the next section, or provide completely different
implementations by extending AbstractMessageChannelMetrics
or AbstractMessageHandlerMetrics
.
See also Micrometer Integration.
In addition to the default metrics factory described earlier, the framework provides the AggregatingMetricsFactory
.
This factory creates AggregatingMessageChannelMetrics
and AggregatingMessageHandlerMetrics
instances.
In very high volume scenarios, the cost of capturing statistics can be prohibitive (the time to make two calls to the system and
store the data for each message).
The aggregating metrics aggregate the response time over a sample of messages.
This can save significant CPU time.
| The statistics are likely to be skewed if messages arrive in bursts. These metrics are intended for use with high, constant-volume, message rates. |
The following example shows how to define an aggregrating metrics factory:
<bean id="aggregatingMetricsFactory"
class="org.springframework.integration.support.management.AggregatingMetricsFactory">
<constructor-arg value="1000" /> <!-- sample size -->
</bean>
The preceding configuration aggregates the duration over 1000 messages. Counts (send and error) are maintained per-message, but the statistics are per 1000 messages.
Customizing the Default Channel and Handler Statistics
See Time-Based Average Estimates and the Javadoc for the ExponentialMovingAverage*
classes for more information about these values.
By default, the DefaultMessageChannelMetrics
and DefaultMessageHandlerMetrics
use a “window” of ten measurements,
a rate period of one second (meaning rate per second) and a decay lapse period of one minute.
If you wish to override these defaults, you can provide a custom MetricsFactory
that returns appropriately configured
metrics and provide a reference to it in the MBean exporter, as described earlier.
The following example shows how to do so:
public static class CustomMetrics implements MetricsFactory {
@Override
public AbstractMessageChannelMetrics createChannelMetrics(String name) {
return new DefaultMessageChannelMetrics(name,
new ExponentialMovingAverage(20, 1000000.),
new ExponentialMovingAverageRate(2000, 120000, 30, true),
new ExponentialMovingAverageRatio(130000, 40, true),
new ExponentialMovingAverageRate(3000, 140000, 50, true));
}
@Override
public AbstractMessageHandlerMetrics createHandlerMetrics(String name) {
return new DefaultMessageHandlerMetrics(name, new ExponentialMovingAverage(20, 1000000.));
}
}
Advanced Customization
The customizations described earlier are wholesale and apply to all appropriate beans exported by the MBean exporter. This is the extent of customization available when you use XML configuration.
Individual beans can be provided with different implementations using by Java @Configuration
or programmatically at
runtime (after the application context has been refreshed) by invoking the configureMetrics
methods on
AbstractMessageChannel
and AbstractMessageHandler
.
Performance Improvement
Previously, the time-based metrics (see Time-Based Average Estimates) were calculated in real time.
The statistics are now calculated when retrieved instead.
This resulted in a significant performance improvement, at the expense of a small amount of additional memory for each statistic.
As discussed earlier, you can disable the statistics altogether while retaining the MBean that allows the invocation of Lifecycle
methods.
12.2. JMX Support
Spring Integration provides channel Adapters for receiving and publishing JMX Notifications.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jmx</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-jmx:5.1.5.RELEASE"
An inbound channel adapter allows for polling JMX MBean attribute values, and an outbound channel adapter allows for invoking JMX MBean operations.
12.2.1. Notification-listening Channel Adapter
The notification-listening channel adapter requires a JMX ObjectName
for the MBean that publishes notifications to which this listener should be registered.
A very simple configuration might resemble the following:
<int-jmx:notification-listening-channel-adapter id="adapter"
channel="channel"
object-name="example.domain:name=publisher"/>
The notification-listening-channel-adapter registers with an MBeanServer at startup, and the default bean name is mbeanServer , which happens to be the same bean name generated when using Spring’s <context:mbean-server/> element.
If you need to use a different name, be sure to include the mbean-server attribute.
|
The adapter can also accept a reference to a NotificationFilter
and a “handback” object to provide some context that is passed back with each notification.
Both of those attributes are optional.
Extending the preceding example to include those attributes as well as an explicit MBeanServer
bean name produces the following example:
<int-jmx:notification-listening-channel-adapter id="adapter"
channel="channel"
mbean-server="someServer"
object-name="example.domain:name=somePublisher"
notification-filter="notificationFilter"
handback="myHandback"/>
The _Notification-listening channel adapter is event-driven and registered with the MBeanServer
directly.
It does not require any poller configuration.
|
For this component only, the
The names of the located MBean(s) are logged when DEBUG level logging is enabled. |
12.2.2. Notification-publishing Channel Adapter
The notification-publishing channel adapter is relatively simple. It requires only a JMX object name in its configuration, as the following example shows:
<context:mbean-export/>
<int-jmx:notification-publishing-channel-adapter id="adapter"
channel="channel"
object-name="example.domain:name=publisher"/>
It also requires that an MBeanExporter
be present in the context.
That is why the <context:mbean-export/>
element is also shown in the preceding example.
When messages are sent to the channel for this adapter, the notification is created from the message content.
If the payload is a String
, it is passed as the message
text for the notification.
Any other payload type is passed as the userData
of the notification.
JMX notifications also have a type
, and it should be a dot-delimited String
.
There are two ways to provide the type
.
Precedence is always given to a message header value associated with the JmxHeaders.NOTIFICATION_TYPE
key.
Alternatively, you can provide a fallback default-notification-type
attribute in the configuration, as the following example shows:
<context:mbean-export/>
<int-jmx:notification-publishing-channel-adapter id="adapter"
channel="channel"
object-name="example.domain:name=publisher"
default-notification-type="some.default.type"/>
12.2.3. Attribute-polling Channel Adapter
The attribute-polling channel adapter is useful when you need to periodically check on some value that is available through an MBean as a managed attribute.
You can configured the poller in the same way as any other polling adapter in Spring Integration (or you can rely on the default poller).
The object-name
and the attribute-name
are required.
An MBeanServer reference is also required.
However, by default, it automatically checks for a bean named mbeanServer
, same as the notification-listening channel adapter described earlier.
The following example shows how to configure an attribute-polling channel adapter with XML:
<int-jmx:attribute-polling-channel-adapter id="adapter"
channel="channel"
object-name="example.domain:name=someService"
attribute-name="InvocationCount">
<int:poller max-messages-per-poll="1" fixed-rate="5000"/>
</int-jmx:attribute-polling-channel-adapter>
12.2.4. Tree-polling Channel Adapter
The tree-polling channel adapter queries the JMX MBean tree and sends a message with a payload that is the graph of objects that matches the query.
By default, the MBeans are mapped to primitives and simple objects, such as Map
, List
, and arrays.
Doing so permits simple transformation to (for example) JSON.
An MBeanServer reference is also required.
However, by default, it automatically checks for a bean named mbeanServer
, same as the notification-listening channel adapter described earlier.
The following example shows how to configure an tree-polling channel adapter with XML:
<int-jmx:tree-polling-channel-adapter id="adapter"
channel="channel"
query-name="example.domain:type=*">
<int:poller max-messages-per-poll="1" fixed-rate="5000"/>
</int-jmx:tree-polling-channel-adapter>
The preceding example includes all of the attributes on the selected MBeans.
You can filter the attributes by providing an MBeanObjectConverter
that has an appropriate filter configured.
You can provide the converter as a reference to a bean definition by using the converter
attribute, or you can use an inner <bean/>
definition.
Spring Integration provides a DefaultMBeanObjectConverter
that can take a MBeanAttributeFilter
in its constructor argument.
Spring Integration provides two standard filters.
The NamedFieldsMBeanAttributeFilter
lets you specify a list of attributes to include.
The NotNamedFieldsMBeanAttributeFilter
lets you specify a list of attributes to exclude.
You can also implement your own filter.
12.2.5. Operation-invoking Channel Adapter
The operation-invoking channel adapter enables message-driven invocation of any managed operation exposed by an MBean. Each invocation requires the operation name to be invoked and the object name of the target MBean. Both of these must be explicitly provided by adapter configuration, as the following example shows:
<int-jmx:operation-invoking-channel-adapter id="adapter"
object-name="example.domain:name=TestBean"
operation-name="ping"/>
Then the adapter only needs to be able to discover the mbeanServer
bean.
If a different bean name is required, then provide the mbean-server
attribute with a reference.
The payload of the message is mapped to the parameters of the operation, if any.
A Map
-typed payload with String
keys is treated as name/value pairs, whereas a List
or array is passed as a simple argument list (with no explicit parameter names).
If the operation requires a single parameter value, the payload can represent that single value.
Also, if the operation requires no parameters, the payload would be ignored.
If you want to expose a channel for a single common operation to be invoked by messages that need not contain headers, that last option works well.
12.2.6. Operation-invoking Outbound Gateway
Similarly to the operation-invoking channel adapter, Spring Integration also provides an operation-invoking outbound gateway, which you can use when dealing with non-void operations when a return value is required.
The return value is sent as the message payload to the reply-channel
specified by the gateway.
The following example shows how to configure an operation-invoking outbound gateway with XML:
<int-jmx:operation-invoking-outbound-gateway request-channel="requestChannel"
reply-channel="replyChannel"
object-name="o.s.i.jmx.config:type=TestBean,name=testBeanGateway"
operation-name="testWithReturn"/>
If you do not provide the reply-channel
attribute, the reply message is sent to the channel identified by the IntegrationMessageHeaderAccessor.REPLY_CHANNEL
header.
That header is typically auto-created by the entry point into a message flow, such as any gateway component.
However, if the message flow was started by manually creating a Spring Integration message and sending it directly to a channel, you must specify the message header explicitly or use the reply-channel
attribute.
12.2.7. MBean Exporter
Spring Integration components may themselvesbe exposed as MBeans when the IntegrationMBeanExporter
is configured.
To create an instance of the IntegrationMBeanExporter
, define a bean and provide a reference to an MBeanServer
and a domain name (if desired).
You can leave out the domain, in which case the default domain is org.springframework.integration
.
The following example shows how to declare an instance of an IntegrationMBeanExporter
and an associated MBeanServer
instance:
<int-jmx:mbean-export id="integrationMBeanExporter"
default-domain="my.company.domain" server="mbeanServer"/>
<bean id="mbeanServer" class="org.springframework.jmx.support.MBeanServerFactoryBean">
<property name="locateExistingServerIfPossible" value="true"/>
</bean>
|
The MBean exporter is orthogonal to the one provided in Spring core.
It registers message channels and message handlers but does not register itself.
You can expose the exporter itself (and certain other components in Spring Integration) by using the standard It also has a useful operation, as discussed in Orderly Shutdown Managed Operation. |
Spring Integration 4.0 introduced the @EnableIntegrationMBeanExport
annotation to allow for convenient configuration of a default integrationMbeanExporter
bean of type IntegrationMBeanExporter
with several useful options at the @Configuration
class level.
The following example shows how to configure this bean:
@Configuration
@EnableIntegration
@EnableIntegrationMBeanExport(server = "mbeanServer", managedComponents = "input")
public class ContextConfiguration {
@Bean
public MBeanServerFactoryBean mbeanServer() {
return new MBeanServerFactoryBean();
}
}
If you need to provide more options or have several IntegrationMBeanExporter
beans (such as
for different MBean Servers or to avoid conflicts with the standard Spring MBeanExporter
— such as through
@EnableMBeanExport
), you can configure an IntegrationMBeanExporter
as a generic bean.
MBean Object Names
All the MessageChannel
, MessageHandler
, and MessageSource
instances in the application are wrapped by the MBean exporter to provide management and monitoring features.
The generated JMX object names for each component type are listed in the following table:
| Component Type | Object Name |
|---|---|
MessageChannel |
`o.s.i:type=MessageChannel,name=<channelName>` |
MessageSource |
`o.s.i:type=MessageSource,name=<channelName>,bean=<source>` |
MessageHandler |
`o.s.i:type=MessageSource,name=<channelName>,bean=<source>` |
The bean
attribute in the object names for sources and handlers takes one of the values in the following table:
| Bean Value | Description |
|---|---|
endpoint |
The bean name of the enclosing endpoint (for example |
anonymous |
An indication that the enclosing endpoint did not have a user-specified bean name, so the JMX name is the input channel name. |
internal |
For well known Spring Integration default components |
handler/source |
None of the above.
Fall back to the |
You can append custom elements to the object name by providing a reference to a Properties
object in the object-name-static-properties
attribute.
Also, since Spring Integration 3.0, you can use a custom ObjectNamingStrategy
by setting the object-naming-strategy
attribute.
Doing so permits greater control over the naming of the MBeans, such as grouping all integration MBeans under an 'Integration' type.
The following example shows one possible custom naming strategy implementation:
public class Namer implements ObjectNamingStrategy {
private final ObjectNamingStrategy realNamer = new KeyNamingStrategy();
@Override
public ObjectName getObjectName(Object managedBean, String beanKey) throws MalformedObjectNameException {
String actualBeanKey = beanKey.replace("type=", "type=Integration,componentType=");
return realNamer.getObjectName(managedBean, actualBeanKey);
}
}
The beanKey
argument is a String
that contain the standard object name, beginning with the default-domain
and including any additional static properties.
The preceding example moves the standard type
part to componentType
and sets the type
to 'Integration', enabling selection of all Integration MBeans in one query:`"my.domain:type=Integration,*`.
Doing so also groups the beans under one tree entry under the domain in such tools as VisualVM.
The default naming strategy is a MetadataNamingStrategy .
The exporter propagates the default-domain to that object to let it generate a fallback object name if parsing of the bean key fails.
If your custom naming strategy is a MetadataNamingStrategy (or a subclass of it), the exporter does not propagate the default-domain .
You must configure it on your strategy bean.
|
Starting with version 5.1; any bean names (represented by the name
key in the object name) will be quoted if they contain any characters that are not allowed in a Java identifier (or period .
).
JMX Improvements
Version 4.2 introduced some important improvements, representing a fairly major overhaul to the JMX support in the framework. These resulted in a significant performance improvement of the JMX statistics collection and much more control thereof. However, it has some implications for user code in a few specific (uncommon) situations. These changes are detailed below, with a caution where necessary.
- Metrics Capture
-
Previously,
MessageSource
,MessageChannel
, andMessageHandler
metrics were captured by wrapping the object in a JDK dynamic proxy to intercept appropriate method calls and capture the statistics. The proxy was added when an integration MBean exporter was declared in the context.Now, the statistics are captured by the beans themselves. See Metrics and Management for more information.
This change means that you no longer automatically get an MBean or statistics for custom MessageHandler
implementations, unless those custom handlers extendAbstractMessageHandler
. The simplest way to resolve this is to extendAbstractMessageHandler
. If you cannot do so, another work around is to implement theMessageHandlerMetrics
interface. For convenience, aDefaultMessageHandlerMetrics
is provided to capture and report statistics. You should invoke thebeforeHandle
andafterHandle
at the appropriate times. YourMessageHandlerMetrics
methods can then delegate to this object to obtain each statistic. Similarly,MessageSource
implementations must extendAbstractMessageSource
or implementMessageSourceMetrics
. Message sources capture only a count, so there is no provided convenience class. You should maintain the count in anAtomicLong
field.The removal of the proxy has two additional benefits:
-
Stack traces in exceptions are reduced (when JMX is enabled) because the proxy is not on the stack
-
Cases where two MBeans were exported for the same bean now only export a single MBean with consolidated attributes and operations (see the MBean consolidation bullet, later).
-
- Resolution
-
System.nanoTime()
(rather thanSystem.currentTimeMillis()
) is now used to capture times . This may provide more accuracy on some JVMs, espcially when you expect durations of less than one millisecond. - Setting Initial Statistics Collection State
-
Previously, when JMX was enabled, all sources, channels, and handlers captured statistics. You can now control whether the statistics are enabled on an individual component. Further, you can capture simple counts on
MessageChannel
instances andMessageHandler
instances instead of capturing the complete time-based statistics. This can have significant performance implications, because you can selectively configure where you need detailed statistics and enable and disable collection at runtime. - @IntegrationManagedResource
-
Similar to the
@ManagedResource
annotation, the@IntegrationManagedResource
marks a class as being eligible to be exported as an MBean. However, it is exported only if the application context has anIntegrationMBeanExporter
.Certain Spring Integration classes (in the
org.springframework.integration
) package) that were previously annotated with`@ManagedResource` are now annotated with both@ManagedResource
and@IntegrationManagedResource
. This is for backwards compatibility (see the next item). Such MBeans are exported by any contextMBeanServer
or by anIntegrationMBeanExporter
(but not both — if both exporters are present, the bean is exported by the integration exporter if the bean matches amanaged-components
pattern). - Consolidated MBeans
-
Certain classes within the framework (mapping routers, for example) have additional attributes and operations over and above those provided by metrics and
Lifecycle
. We use aRouter
as an example here.Previously, beans of these types were exported as two distinct MBeans:
-
The metrics MBean (with an object name such as
intDomain:type=MessageHandler,name=myRouter,bean=endpoint
). This MBean had metrics attributes and metrics/Lifecycle operations. -
A second MBean (with an object name such as
ctxDomain:name=org.springframework.integration.config.
RouterFactoryBean#0
,type=MethodInvokingRouter`) was exported with the channel mappings attribute and operations.Now the attributes and operations are consolidated into a single MBean. The object name depends on the exporter. If exported by the integration MBean exporter, the object name is, for example:
intDomain:type=MessageHandler,name=myRouter,bean=endpoint
. If exported by another exporter, the object name is, for example:ctxDomain:name=org.springframework.integration.config.
RouterFactoryBean#0,type=MethodInvokingRouter
. There is no difference between these MBeans (aside from the object name), except that the statistics are not enabled (the attributes are0
) by exporters other than the integration exporter. You can enable statistics at runtime by using the JMX operations. When exported by the integration MBean exporter, the initial state can be managed as described earlier.If you currently use the second MBean to change, for example, channel mappings and you use the integration MBean exporter, note that the object name has changed because of the MBean consolidation. There is no change if you are not using the integration MBean exporter.
-
- MBean Exporter Bean Name Patterns
-
Previously, the
managed-components
patterns were inclusive only. If a bean name matched one of the patterns, it would be included. Now, the pattern can be negated by prefixing it with!
. For example,!thing*, things
matches all bean names that do not start withthing
exceptthings
. Patterns are evaluated left to right. The first match (positive or negative) wins, and then no further patterns are applied.The addition of this syntax to the pattern causes one possible (although perhaps unlikely) problem. If you have a bean named "!thing"
and you included a pattern of!thing
in your MBean exporter’smanaged-components
patterns, it no longer matches; the pattern now matches all beans not namedthing
. In this case, you can escape the!
in the pattern with\
. The\!thing
pattern matches a bean named!thing
. - IntegrationMBeanExporter changes
-
The
IntegrationMBeanExporter
no longer implementsSmartLifecycle
. This means thatstart()
andstop()
operations are no longer available to registerand unregister MBeans. The MBeans are now registered during context initialization and unregistered when the context is destroyed.
Orderly Shutdown Managed Operation
The MBean exporter provides a JMX operation to shut down the application in an orderly manner, intended for use before terminating the JVM. The following example shows how to use it:
public void stopActiveComponents(long howLong)
Its use and operation are described in Orderly Shutdown.
12.3. Message History
The key benefit of a messaging architecture is loose coupling such that participating components do not maintain any awareness about one another. This fact alone makes an application extremely flexible, letting you change components without affecting the rest of the flow, change messaging routes, change message consuming styles (polling versus event driven), and so on. However, this unassuming style of architecture could prove to be difficult when things go wrong. When debugging, you probably want as much information (its origin, the channels it has traversed, and other details) about the message as you can get.
Message history is one of those patterns that helps by giving you an option to maintain some level of awareness of a message path either for debugging purposes or for maintaining an audit trail. Spring integration provides a simple way to configure your message flows to maintain the message history by adding a header to the message and updating that header every time a message passes through a tracked component.
12.3.1. Message History Configuration
To enable message history, you need only define the message-history
element in your configuration, as shown in the following example:
<int:message-history/>
Now every named component (component that has an 'id' defined) is tracked.
The framework sets the 'history' header in your message.
Its value a List<Properties>
.
Consider the following configuration example:
<int:gateway id="sampleGateway"
service-interface="org.springframework.integration.history.sample.SampleGateway"
default-request-channel="bridgeInChannel"/>
<int:chain id="sampleChain" input-channel="chainChannel" output-channel="filterChannel">
<int:header-enricher>
<int:header name="baz" value="baz"/>
</int:header-enricher>
</int:chain>
The preceding configuration produces a simple message history structure, with output similar to the following:
[{name=sampleGateway, type=gateway, timestamp=1283281668091},
{name=sampleChain, type=chain, timestamp=1283281668094}]
To get access to message history, you need only access the MessageHistory
header.
The folloiwng example shows how to do so:
Iterator<Properties> historyIterator =
message.getHeaders().get(MessageHistory.HEADER_NAME, MessageHistory.class).iterator();
assertTrue(historyIterator.hasNext());
Properties gatewayHistory = historyIterator.next();
assertEquals("sampleGateway", gatewayHistory.get("name"));
assertTrue(historyIterator.hasNext());
Properties chainHistory = historyIterator.next();
assertEquals("sampleChain", chainHistory.get("name"));
You might not want to track all of the components.
To limit the history to certain components based on their names, you can provide the tracked-components
attribute and specify a comma-delimited list of component names and patterns that match the components you want to track.
The following example shows how to do so:
<int:message-history tracked-components="*Gateway, sample*, aName"/>
In the preceding example, message history is maintained only for the components that end with 'Gateway', start with 'sample', or match the name, 'aName', exactly.
Starting with version 4.0, you can also use the @EnableMessageHistory
annotation in a @Configuration
class.
In addition, the MessageHistoryConfigurer
bean is now exposed as a JMX MBean by the IntegrationMBeanExporter
(see MBean Exporter), letting you change the patterns at runtime.
Note, however, that the bean must be stopped (turning off message history) in order to change the patterns.
This feature might be useful to temporarily turn on history to analyze a system.
The MBean’s object name is <domain>:name=messageHistoryConfigurer,type=MessageHistoryConfigurer
.
If multiple beans (declared by @EnableMessageHistory and <message-history/> ) exist, they must all have identical component name patterns (when trimmed and sorted).
Do not use a generic <bean/> definition for the MessageHistoryConfigurer .
|
| By definition, the message history header is immutable (you cannot re-write history). Therefore, when writing message history values, the components either create new messages (when the component is an origin) or they copy the history from a request message, modifying it and setting the new list on a reply message. In either case, the values can be appended even if the message itself is crossing thread boundaries. That means that the history values can greatly simplify debugging in an asynchronous message flow. |
12.4. Message Store
The Enterprise Integration Patterns (EIP) book identifies several patterns that have the ability to buffer messages.
For example, an aggregator buffers messages until they can be released, and a QueueChannel
buffers messages until consumers explicitly receive those messages from that channel.
Because of the failures that can occur at any point within your message flow, EIP components that buffer messages also introduce a point where messages could be lost.
To mitigate the risk of losing messages, EIP defines the message store pattern, which lets EIP components store messages, typically in some type of persistent store (such as an RDBMS).
Spring Integration provides support for the message store pattern by:
-
Defining an
org.springframework.integration.store.MessageStore
strategy interface -
Providing several implementations of this interface
-
Exposing a
message-store
attribute on all components that have the capability to buffer messages so that you can inject any instance that implements theMessageStore
interface.
Details on how to configure a specific message store implementation and how to inject a MessageStore
implementation into a specific buffering component are described throughout the manual (see the specific component, such as QueueChannel, Aggregator, Delayer, and others).
The following pair of examples show how to add a reference to a message store for a QueueChannel
and for an aggregator:
<int:channel id="myQueueChannel">
<int:queue message-store="refToMessageStore"/>
<int:channel>
<int:aggregator … message-store="refToMessageStore"/>
By default, messages are stored in-memory by using o.s.i.store.SimpleMessageStore
, an implementation of MessageStore
.
That might be fine for development or simple low-volume environments where the potential loss of non-persistent messages is not a concern.
However, the typical production application needs a more robust option, not only to mitigate the risk of message loss but also to avoid potential out-of-memory errors.
Therefore, we also provide MessageStore
implementations for a variety of data-stores.
The following is a complete list of supported implementations:
-
JDBC Message Store: Uses an RDBMS to store messages
-
Redis Message Store: Uses a Redis key/value datastore to store messages
-
MongoDB Message Store: Uses a MongoDB document store to store messages
-
Gemfire Message Store: Uses a Gemfire distributed cache to store messages
|
However, be aware of some limitations while using persistent implementations of the The Message data (payload and headers) is serialized and deserialized by using different serialization strategies, depending on the implementation of the Pay special attention to the headers that represent certain types of data.
For example, if one of the headers contains an instance of some Spring bean, upon deserialization, you may end up with a different instance of that bean, which directly affects some of the implicit headers created by the framework (such as Beginning with Spring Integration version 3.0, you can resolve this issue with a header enricher configured to replace these headers with a name after registering the channel with the Also, consider what happens when you configure a message-flow as follows: gateway → queue-channel (backed by a persistent Message Store) → service-activator.
That gateway creates a temporary reply channel, which is lost by the time the service-activator’s poller reads from the queue.
Again, you can use the header enricher to replace the headers with a For more information, see Header Enricher. |
Spring Integration 4.0 introduced two new interfaces:
-
ChannelMessageStore
: To implement operations specific forQueueChannel
instances -
PriorityCapableChannelMessageStore
: To markMessageStore
implementations to be used forPriorityChannel
instances and to provide priority order for persisted messages.
The real behavior depends on the implementation.
The framework provides the following implementations, which can be used as a persistent MessageStore
for QueueChannel
and PriorityChannel
:
|
Caution about
SimpleMessageStore Starting with version 4.1, the Users accessing the group store outside of components such as aggregators now get a direct reference to the group being used by the aggregator instead of a copy. Manipulation of the group outside of the aggregator may cause unpredictable results. For this reason, you should either not perform such manipulation or set the |
12.4.1. Using MessageGroupFactory
Starting with version 4.3, some MessageGroupStore
implementations can be injected with a custom
MessageGroupFactory
strategy to create and customize the MessageGroup
instances used by the MessageGroupStore
.
This defaults to a SimpleMessageGroupFactory
, which produces SimpleMessageGroup
instances based on the GroupType.HASH_SET
(LinkedHashSet
) internal collection.
Other possible options are SYNCHRONISED_SET
and BLOCKING_QUEUE
, where the last one can be used to reinstate the
previous SimpleMessageGroup
behavior.
Also the PERSISTENT
option is available.
See the next section for more information.
Starting with version 5.0.1, the LIST
option is also available for when the order and uniqueness of messages in the group does not matter.
12.4.2. Persistent MessageGroupStore
and Lazy-load
Starting with version 4.3, all persistent MessageGroupStore
instances retrieve MessageGroup
instances and their messages
from the store in the lazy-load manner.
In most cases, it is useful for the correlation MessageHandler
instances (see Aggregator and Resequencer),
when it would add overhead to load entire the MessageGroup
from the store on each correlation operation.
You can use the AbstractMessageGroupStore.setLazyLoadMessageGroups(false)
option to switch off the lazy-load behavior from the configuration.
Our performance tests for lazy-load on MongoDB MessageStore
(MongoDB Message Store) and
<aggregator>
(Aggregator)
use a custom release-strategy
similar to the following:
<int:aggregator input-channel="inputChannel"
output-channel="outputChannel"
message-store="mongoStore"
release-strategy-expression="size() == 1000"/>
It produces results similar to the following for 1000 simple messages:
... StopWatch 'Lazy-Load Performance': running time (millis) = 38918 ----------------------------------------- ms % Task name ----------------------------------------- 02652 007% Lazy-Load 36266 093% Eager ...
12.5. Metadata Store
Many external systems, services, or resources are not transactional (Twitter, RSS, file systems, and so on), and there is no any ability to mark the data as read.
Also, sometimes, you may need to implement the Enterprise Integration Pattern idempotent receiver in some integration solutions.
To achieve this goal and store some previous state of the endpoint before the next interaction with external system or to deal with the next message, Spring Integration provides the metadata store component as an an implementation of the org.springframework.integration.metadata.MetadataStore
interface with a general key-value contract.
The metadata store is designed to store various types of generic metadata (for example, the published date of the last feed entry that has been processed) to help components such as the feed adapter deal with duplicates.
If a component is not directly provided with a reference to a MetadataStore
, the algorithm for locating a metadata store is as follows: First, look for a bean with a metadataStore
ID in the application context.
If one is found, use it.
Otherwise, create a new instance of SimpleMetadataStore
, which is an in-memory implementation that persists only metadata within the lifecycle of the currently running application context.
This means that, upon restart, you may end up with duplicate entries.
If you need to persist metadata between application context restarts, the framework provides the following persistent MetadataStores
:
-
PropertiesPersistingMetadataStore
The PropertiesPersistingMetadataStore
is backed by a properties file and a PropertiesPersister
.
By default, it persists only the state when the application context is closed normally.
It implements Flushable
so that you can persist the state at will, by invoking flush()
.
The following example shows how to configure a 'PropertiesPersistingMetadataStore' with XML:
<bean id="metadataStore"
class="org.springframework.integration.metadata.PropertiesPersistingMetadataStore"/>
Alternatively, you can provide your own implementation of the MetadataStore
interface (for example,
JdbcMetadataStore
) and configure it as a bean in the application context.
Starting with version 4.0, SimpleMetadataStore
, PropertiesPersistingMetadataStore
, and RedisMetadataStore
implement ConcurrentMetadataStore
.
These provide for atomic updates and can be used across multiple component or application instances.
12.5.1. Idempotent Receiver and Metadata Store
The metadata store is useful for implementing the EIP idempotent receiver pattern when there is need to filter an incoming message if it has already been processed and you can discard it or perform some other logic on discarding. The following configuration shows an example of how to do so:
<int:filter input-channel="serviceChannel"
output-channel="idempotentServiceChannel"
discard-channel="discardChannel"
expression="@metadataStore.get(headers.businessKey) == null"/>
<int:publish-subscribe-channel id="idempotentServiceChannel"/>
<int:outbound-channel-adapter channel="idempotentServiceChannel"
expression="@metadataStore.put(headers.businessKey, '')"/>
<int:service-activator input-channel="idempotentServiceChannel" ref="service"/>
The value
of the idempotent entry may be an expiration date, after which that entry should be removed from metadata store by some scheduled reaper.
12.5.2. MetadataStoreListener
Some metadata stores (currently only zookeeper) support registering a listener to receive events when items change, as the following example shows:
public interface MetadataStoreListener {
void onAdd(String key, String value);
void onRemove(String key, String oldValue);
void onUpdate(String key, String newValue);
}
See the Javadoc for more information.
The MetadataStoreListenerAdapter
can be subclassed if you are interested only in a subset of events.
12.6. Control Bus
As described in the Enterprise Integration Patterns (EIP) book, the idea behind the control bus is that the same messaging system can be used for monitoring and managing the components within the framework as is used for “application-level” messaging. In Spring Integration, we build upon the adapters described above so that you can send messages as a means of invoking exposed operations.
The following example shows how to configure a control bus with XML:
<int:control-bus input-channel="operationChannel"/>
The control bus has an input channel that can be accessed for invoking operations on the beans in the application context. It also has all the common properties of a service activating endpoint. For example, you can specify an output channel if the result of the operation has a return value that you want to send on to a downstream channel.
The control bus runs messages on the input channel as Spring Expression Language (SpEL) expressions.
It takes a message, compiles the body to an expression, adds some context, and then runs it.
The default context supports any method that has been annotated with @ManagedAttribute
or @ManagedOperation
.
It also supports the methods on Spring’s Lifecycle
interface, and it supports methods that are used to configure several of Spring’s TaskExecutor
and TaskScheduler
implementations.
The simplest way to ensure that your own methods are available to the control bus is to use the @ManagedAttribute
or @ManagedOperation
annotations.
Since those annotations are also used for exposing methods to a JMX MBean registry, they offer a convenient by-product: Often, the same types of operations you want to expose to the control bus are reasonable for exposing through JMX).
Resolution of any particular instance within the application context is achieved in the typical SpEL syntax.
To do so, provide the bean name with the SpEL prefix for beans (@
).
For example, to execute a method on a Spring Bean, a client could send a message to the operation channel as follows:
Message operation = MessageBuilder.withPayload("@myServiceBean.shutdown()").build();
operationChannel.send(operation)
The root of the context for the expression is the Message
itself, so you also have access to the payload
and headers
as variables within your expression.
This is consistent with all the other expression support in Spring Integration endpoints.
With Java annotations, you can configured the control bus as follows:
@Bean
@ServiceActivator(inputChannel = "operationChannel")
public ExpressionControlBusFactoryBean controlBus() {
return new ExpressionControlBusFactoryBean();
}
Similarly, you can configure Java DSL flow definitions as follows:
@Bean
public IntegrationFlow controlBusFlow() {
return IntegrationFlows.from("controlBus")
.controlBus()
.get();
}
If you prefer to use lambdas with automatic DirectChannel
creation, you can create a control bus as follows:
@Bean
public IntegrationFlow controlBus() {
return IntegrationFlowDefinition::controlBus;
}
In this case, the channel is named controlBus.input
.
12.7. Orderly Shutdown
As described in "MBean Exporter", the MBean exporter provides a JMX operation called stopActiveComponents
, which is used to stop the application in an orderly manner.
The operation has a single Long
parameter.
The parameter indicates how long (in milliseconds) the operation waits to allow in-flight messages to complete.
The operation works as follows:
-
Call
beforeShutdown()
on all beans that implementOrderlyShutdownCapable
.Doing so lets such components prepare for shutdown. Examples of components that implement this interface and what they do with this call include JMS and AMQP message-driven adapters that stop their listener containers, TCP server connection factories that stop accepting new connections (while keeping existing connections open), TCP inbound endpoints that drop (log) any new messages received, and HTTP inbound endpoints that return
503 - Service Unavailable
for any new requests. -
Stop any active channels, such as JMS- or AMQP-backed channels.
-
Stop all
MessageSource
instances. -
Stop all inbound
MessageProducer
s (that are notOrderlyShutdownCapable
). -
Wait for any remaining time left, as defined by the value of the
Long
parameter passed in to the operation.Doing so lets any in-flight messages complete their journeys. It is therefore important to select an appropriate timeout when invoking this operation.
-
Call
afterShutdown()
on allOrderlyShutdownCapable
components.Doing so lets such components perform final shutdown tasks (closing all open sockets, for example).
As discussed in Orderly Shutdown Managed Operation, this operation can be invoked by using JMX.
If you wish to programmatically invoke the method, you need to inject or otherwise get a reference to the IntegrationMBeanExporter
.
If no id
attribute is provided on the <int-jmx:mbean-export/>
definition, the bean has a generated name.
This name contains a random component to avoid ObjectName
collisions if multiple Spring Integration contexts exist in the same JVM (MBeanServer
).
For this reason, if you wish to invoke the method programmatically, we recommend that you provide the exporter with an id
attribute so that you can easily access it in the application context.
Finally, the operation can be invoked by using the <control-bus>
element.
See the monitoring Spring Integration sample application for details.
The algorithm described earlier was improved in version 4.1.
Previously, all task executors and schedulers were stopped.
This could cause mid-flow messages in QueueChannel instances to remain.
Now the shutdown leaves pollers running, to let these messages be drained and processed.
|
12.8. Integration Graph
Starting with version 4.3, Spring Integration provides access to an application’s runtime object model, which can, optionally, include component metrics.
It is exposed as a graph, which may be used to visualize the current state of the integration application.
The o.s.i.support.management.graph
package contains all the required classes to collect, build, and render the runtime state of Spring Integration components as a single tree-like Graph
object.
The IntegrationGraphServer
should be declared as a bean to build, retrieve, and refresh the Graph
object.
The resulting Graph
object can be serialized to any format, although JSON is flexible and convenient to parse and represent on the client side.
A Spring Integration application with only the default components would expose a graph as follows:
{
"contentDescriptor": {
"providerVersion": "4.3.0.RELEASE",
"providerFormatVersion": 1.0,
"provider": "spring-integration",
"name": "myApplication"
},
"nodes": [
{
"nodeId": 1,
"name": "nullChannel",
"stats": null,
"componentType": "channel"
},
{
"nodeId": 2,
"name": "errorChannel",
"stats": null,
"componentType": "publish-subscribe-channel"
},
{
"nodeId": 3,
"name": "_org.springframework.integration.errorLogger",
"stats": {
"duration": {
"count": 0,
"min": 0.0,
"max": 0.0,
"mean": 0.0,
"standardDeviation": 0.0,
"countLong": 0
},
"errorCount": 0,
"standardDeviationDuration": 0.0,
"countsEnabled": true,
"statsEnabled": true,
"loggingEnabled": false,
"handleCount": 0,
"meanDuration": 0.0,
"maxDuration": 0.0,
"minDuration": 0.0,
"activeCount": 0
},
"componentType": "logging-channel-adapter",
"output": null,
"input": "errorChannel"
}
],
"links": [
{
"from": 2,
"to": 3,
"type": "input"
}
]
}
In the preceding example, the graph consists of three top-level elements.
The contentDescriptor
graph element contains general information about the application providing the data.
The name
can be customized on the IntegrationGraphServer
bean or in the spring.application.name
application context environment property.
Other properties are provided by the framework and let you distinguish a similar model from other sources.
The links
graph element represents connections between nodes from the nodes
graph element and, therefore, between integration components in the source Spring Integration application.
For example, from a MessageChannel
to an EventDrivenConsumer
with some MessageHandler
or from an AbstractReplyProducingMessageHandler
to a MessageChannel
.
For convenience and to let you determine a link’s purpose, the model includes the type
attribute.
The possible types are:
-
input
: Identifies the direction fromMessageChannel
to the endpoint,inputChannel
, orrequestChannel
property -
output
: The direction from theMessageHandler
,MessageProducer
, orSourcePollingChannelAdapter
to theMessageChannel
through anoutputChannel
orreplyChannel
property -
error
: FromMessageHandler
onPollingConsumer
orMessageProducer
orSourcePollingChannelAdapter
to theMessageChannel
through anerrorChannel
property; -
discard
: FromDiscardingMessageHandler
(such asMessageFilter
) to theMessageChannel
through anerrorChannel
property. -
route
: FromAbstractMappingMessageRouter
(such asHeaderValueRouter
) to theMessageChannel
. Similar tooutput
but determined at run-time. May be a configured channel mapping or a dynamically resolved channel. Routers typically retain only up to 100 dynamic routes for this purpose, but you can modify this value by setting thedynamicChannelLimit
property.
The information from this element can be used by a visualization tool to render connections between nodes from the nodes
graph element, where the from
and to
numbers represent the value from the nodeId
property of the linked nodes.
For example, the link
element can be used to determine the proper port
on the target node.
The following “text image” shows the relationships between the types:
+---(discard) | +----o----+ | | | | | | (input)--o o---(output) | | | | | | +----o----+ | +---(error)
The nodes
graph element is perhaps the most interesting, because its elements contain not only the runtime components with their componentType
instances and name
values but can also optionally contain metrics exposed by the component.
Node elements contain various properties that are generally self-explanatory.
For example, expression-based components include the expression
property that contains the primary expression string for the component.
To enable the metrics, add an @EnableIntegrationManagement
to a @Configuration
class or add an <int:management/>
element to your XML configuration.
You can control exactly which components in the framework collect statistics.
See Metrics and Management for complete information.
See the stats
attribute from the o.s.i.errorLogger
component in the JSON example shown earlier.
In this case, The nullChannel
and errorChannel
do not provide statistics information, because the configuration for this example was as follows:
@Configuration
@EnableIntegration
@EnableIntegrationManagement(statsEnabled = "_org.springframework.integration.errorLogger.handler",
countsEnabled = "!*",
defaultLoggingEnabled = "false")
public class ManagementConfiguration {
@Bean
public IntegrationGraphServer integrationGraphServer() {
return new IntegrationGraphServer();
}
}
The nodeId
represents a unique incremental identifier to let you distinguish one component from another.
It is also used in the links
element to represent a relationship (connection) of this component to others, if any.
The input
and output
attributes are for the inputChannel
and outputChannel
properties of the AbstractEndpoint
, MessageHandler
, SourcePollingChannelAdapter
, or MessageProducerSupport
.
See the next section for more information.
Starting with version 5.1, the IntegrationGraphServer
accepts a Function<NamedComponent, Map<String, Object>> additionalPropertiesCallback
for population of additional properties on the IntegrationNode
for a particular NamedComponent
.
For example you can expose the SmartLifecycle
autoStartup
and running
properties into the target graph:
server.setAdditionalPropertiesCallback(namedComponent -> {
Map<String, Object> properties = null;
if (namedComponent instanceof SmartLifecycle) {
SmartLifecycle smartLifecycle = (SmartLifecycle) namedComponent;
properties = new HashMap<>();
properties.put("auto-startup", smartLifecycle.isAutoStartup());
properties.put("running", smartLifecycle.isRunning());
}
return properties;
});
12.8.1. Graph Runtime Model
Spring Integration components have various levels of complexity.
For example, any polled MessageSource
also has a SourcePollingChannelAdapter
and a MessageChannel
to which to periodically send messages from the source data.
Other components might be middleware request-reply components (such as JmsOutboundGateway
) with a consuming AbstractEndpoint
to subscribe to (or poll) the requestChannel
(input
) for messages, and a replyChannel
(output
) to produce a reply message to send downstream.
Meanwhile, any MessageProducerSupport
implementation (such as ApplicationEventListeningMessageProducer
) wraps some source protocol listening logic and sends messages to the outputChannel
.
Within the graph, Spring Integration components are represented by using the IntegrationNode
class hierarchy, which you can find in the o.s.i.support.management.graph
package.
For example, you can use the ErrorCapableDiscardingMessageHandlerNode
for the AggregatingMessageHandler
(because it has a discardChannel
option) and can produce errors when consuming from a PollableChannel
by using a PollingConsumer
.
Another example is CompositeMessageHandlerNode
— for a MessageHandlerChain
when subscribed to a SubscribableChannel
by using an EventDrivenConsumer
.
The @MessagingGateway (see Messaging Gateways) provides nodes for each of its method, where the name attribute is based on the gateway’s bean name and the short method signature.
Consider the following example of a gateway:
|
@MessagingGateway(defaultRequestChannel = "four")
public interface Gate {
void foo(String foo);
void foo(Integer foo);
void bar(String bar);
}
The preceding gateway produces nodes similar to the following:
{
"nodeId" : 10,
"name" : "gate.bar(class java.lang.String)",
"stats" : null,
"componentType" : "gateway",
"output" : "four",
"errors" : null
},
{
"nodeId" : 11,
"name" : "gate.foo(class java.lang.String)",
"stats" : null,
"componentType" : "gateway",
"output" : "four",
"errors" : null
},
{
"nodeId" : 12,
"name" : "gate.foo(class java.lang.Integer)",
"stats" : null,
"componentType" : "gateway",
"output" : "four",
"errors" : null
}
You can use this IntegrationNode
hierarchy for parsing the graph model on the client side as well as to understand the general Spring Integration runtime behavior.
See also Programming Tips and Tricks for more information.
12.9. Integration Graph Controller
If your application is web-based (or built on top of Spring Boot with an embedded web container) and the Spring Integration HTTP or WebFlux module (see HTTP Support and WebFlux Support, respectively) is present on the classpath, you can use a IntegrationGraphController
to expose the IntegrationGraphServer
functionality as a REST service.
For this purpose, the @EnableIntegrationGraphController
and @Configuration
class annotations and the <int-http:graph-controller/>
XML element are available in the HTTP module.
Together with the @EnableWebMvc
annotation (or <mvc:annotation-driven/>
for XML definitions), this configuration registers an IntegrationGraphController
@RestController
where its @RequestMapping.path
can be configured on the @EnableIntegrationGraphController
annotation or <int-http:graph-controller/>
element.
The default path is /integration
.
The IntegrationGraphController
@RestController
provides the following services:
-
@GetMapping(name = "getGraph")
: To retrieve the state of the Spring Integration components since the lastIntegrationGraphServer
refresh. Theo.s.i.support.management.graph.Graph
is returned as a@ResponseBody
of the REST service. -
@GetMapping(path = "/refresh", name = "refreshGraph")
: To refresh the currentGraph
for the actual runtime state and return it as a REST response. It is not necessary to refresh the graph for metrics. They are provided in real-time when the graph is retrieved. Refresh can be called if the application context has been modified since the graph was last retrieved. In that case, the graph is completely rebuilt.
You can set security and cross-origin restrictions for the IntegrationGraphController
with the standard configuration options and components provided by the Spring Security and Spring MVC projects.
The following example achieves those goals:
<mvc:annotation-driven />
<mvc:cors>
<mvc:mapping path="/myIntegration/**"
allowed-origins="http://localhost:9090"
allowed-methods="GET" />
</mvc:cors>
<security:http>
<security:intercept-url pattern="/myIntegration/**" access="ROLE_ADMIN" />
</security:http>
<int-http:graph-controller path="/myIntegration" />
The following example shows how to do the same thing with Java configuration:
@Configuration
@EnableWebMvc // or @EnableWebFlux
@EnableWebSecurity // or @EnableWebFluxSecurity
@EnableIntegration
@EnableIntegrationGraphController(path = "/testIntegration", allowedOrigins="http://localhost:9090")
public class IntegrationConfiguration extends WebSecurityConfigurerAdapter {
@Override
protected void configure(HttpSecurity http) throws Exception {
http
.authorizeRequests()
.antMatchers("/testIntegration/**").hasRole("ADMIN")
// ...
.formLogin();
}
//...
}
Note that, for convenience, the @EnableIntegrationGraphController
annotation provides an allowedOrigins
attribute.
This provides GET
access to the path
.
For more sophistication, you can configure the CORS mappings by using standard Spring MVC mechanisms.
Integration Endpoints
This section covers the various channel adapters and messaging gateways provided by Spring Integration to support message-based communication with external systems.
Each system, from AMQP to Zookeeper, has its own integration requirements, and this section covers them.
13. Endpoint Quick Reference Table
As discussed in the earlier sections, Spring Integration provides a number of endpoints used to interface with external systems, file systems, and others.
For transparent dependency management Spring Integration provides a bill-of-materials POM to be imported into the Maven configuration:
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-bom</artifactId>
<version>5.1.5.RELEASE</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
To recap:
-
Inbound channel adapters are used for one-way integration to bring data into the messaging application.
-
Outbound channel adapters are used for one-way integration to send data out of the messaging application.
-
Inbound gateways are used for a bidirectional integration flow, where some other system invokes the messaging application and receives a reply.
-
Outbound Gateways are used for a bidirectional integration flow, where the messaging application invokes some external service or entity and expects a result.
The following table summarizes the various endpoints with quick links to the appropriate chapter.
In addition, as discussed in Core Messaging, Spring Integration provides endpoints for interfacing with Plain Old Java Objects (POJOs).
As discussed in Channel Adapter, the <int:inbound-channel-adapter>
element lets you poll a Java method for data.
The <int:outbound-channel-adapter>
element lets you send data to a void
method.
As discussed in Messaging Gateways, the <int:gateway>
element lets any Java program invoke a messaging flow.
Each of these works without requiring any source-level dependencies on Spring Integration.
The equivalent of an outbound gateway in this context is using a service activator (see Service Activator) to invoke a method that returns an Object
of some kind.
14. AMQP Support
Spring Integration provides channel adapters for receiving and sending messages by using the Advanced Message Queuing Protocol (AMQP).
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-amqp</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-amqp:5.1.5.RELEASE"
The following adapters are available:
Spring Integration also provides a point-to-point message channel and a publish-subscribe message channel backed by AMQP Exchanges and Queues.
To provide AMQP support, Spring Integration relies on (Spring AMQP), which applies core Spring concepts to the development of AMQP-based messaging solutions. Spring AMQP provides similar semantics to (Spring JMS).
Whereas the provided AMQP Channel Adapters are intended for unidirectional messaging (send or receive) only, Spring Integration also provides inbound and outbound AMQP gateways for request-reply operations.
TIP: You should familiarize yourself with the reference documentation of the Spring AMQP project. It provides much more in-depth information about Spring’s integration with AMQP in general and RabbitMQ in particular.
14.1. Inbound Channel Adapter
The following listing shows the possible configuration options for an AMQP Inbound Channel Adapter:
<int-amqp:inbound-channel-adapter
id="inboundAmqp" (1)
channel="inboundChannel" (2)
queue-names="si.test.queue" (3)
acknowledge-mode="AUTO" (4)
advice-chain="" (5)
channel-transacted="" (6)
concurrent-consumers="" (7)
connection-factory="" (8)
error-channel="" (9)
expose-listener-channel="" (10)
header-mapper="" (11)
mapped-request-headers="" (12)
listener-container="" (13)
message-converter="" (14)
message-properties-converter="" (15)
phase="" (16)
prefetch-count="" (17)
receive-timeout="" (18)
recovery-interval="" (19)
missing-queues-fatal="" (20)
shutdown-timeout="" (21)
task-executor="" (22)
transaction-attribute="" (23)
transaction-manager="" (24)
tx-size="" (25)
consumers-per-queue /> (26)
| 1 | The unique ID for this adapter. Optional. |
| 2 | Message channel to which converted messages should be sent. Required. |
| 3 | Names of the AMQP queues (comma-separated list) from which messages should be consumed. Required. |
| 4 | Acknowledge mode for the MessageListenerContainer .
When set to MANUAL , the delivery tag and channel are provided in message headers amqp_deliveryTag and amqp_channel , respectively.
The user application is responsible for acknowledgement.
NONE means no acknowledgements (autoAck ).
AUTO means the adapter’s container acknowledges when the downstream flow completes.
Optional (defaults to AUTO).
See Inbound Endpoint Acknowledge Mode. |
| 5 | Extra AOP Advices to handle cross-cutting behavior associated with this inbound channel adapter. Optional. |
| 6 | Flag to indicate that channels created by this component are transactional. If true, it tells the framework to use a transactional channel and to end all operations (send or receive) with a commit or rollback, depending on the outcome, with an exception that signals a rollback. Optional (Defaults to false). |
| 7 | Specify the number of concurrent consumers to create.
The default is 1 .
We recommend raising the number of concurrent consumers to scale the consumption of messages coming in from a queue.
However, note that any ordering guarantees are lost once multiple consumers are registered.
In general, use one consumer for low-volume queues.
Not allowed when 'consumers-per-queue' is set.
Optional. |
| 8 | Bean reference to the RabbitMQ ConnectionFactory .
Optional (defaults to 'connectionFactory'). |
| 9 | Message channel to which error messages should be sent. Optional. |
| 10 | Whether the listener channel (com.rabbitmq.client.Channel) is exposed to a registered ChannelAwareMessageListener .
Optional (defaults to true). |
| 11 | A reference to an AmqpHeaderMapper to use when receiving AMQP Messages.
Optional.
By default, only standard AMQP properties (such as contentType ) are copied to Spring Integration MessageHeaders .
Any user-defined headers within the AMQP MessageProperties are NOT copied to the message by the default DefaultAmqpHeaderMapper .
Not allowed if 'request-header-names' is provided. |
| 12 | Comma-separated list of the names of AMQP Headers to be mapped from the AMQP request into the MessageHeaders .
This can only be provided if the 'header-mapper' reference is not provided.
The values in this list can also be simple patterns to be matched against the header names (such as "*" or "thing1*, thing2" or "*something"). |
| 13 | Reference to the AbstractMessageListenerContainer to use for receiving AMQP Messages.
If this attribute is provided, no other attribute related to the listener container configuration should be provided.
In other words, by setting this reference, you must take full responsibility for the listener container configuration.
The only exception is the MessageListener itself.
Since that is actually the core responsibility of this channel adapter implementation, the referenced listener container must not already have its own MessageListener .
Optional. |
| 14 | The MessageConverter to use when receiving AMQP messages.
Optional. |
| 15 | The MessagePropertiesConverter to use when receiving AMQP messages.
Optional. |
| 16 | Specifies the phase in which the underlying AbstractMessageListenerContainer should be started and stopped.
The startup order proceeds from lowest to highest, and the shutdown order is the reverse of that.
By default, this value is Integer.MAX_VALUE , meaning that this container starts as late as possible and stops as soon as possible.
Optional. |
| 17 | Tells the AMQP broker how many messages to send to each consumer in a single request.
Often, you can set this value high to improve throughput.
It should be greater than or equal to the transaction size (see the tx-size attribute, later in this list).
Optional (defaults to 1 ). |
| 18 | Receive timeout in milliseconds.
Optional (defaults to 1000 ). |
| 19 | Specifies the interval between recovery attempts of the underlying AbstractMessageListenerContainer (in milliseconds).
Optional (defaults to 5000 ). |
| 20 | If 'true' and none of the queues are available on the broker, the container throws a fatal exception during startup and stops if the queues are deleted when the container is running (after making three attempts to passively declare the queues).
If false , the container does not throw an exception and goes 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 AbstractMessageListenerContainer is stopped and before the AMQP connection is forced closed.
If any workers are active when the shutdown signal comes, they are allowed to finish processing as long as they can finish within this timeout.
Otherwise, the connection is closed and messages remain unacknowledged (if the channel is transactional).
Optional (defaults to 5000 ). |
| 22 | By default, the underlying AbstractMessageListenerContainer uses a SimpleAsyncTaskExecutor implementation, that fires up a new thread for each task, running it asynchronously.
By default, the number of concurrent threads is unlimited.
Note that this implementation does not reuse threads.
Consider using a thread-pooling TaskExecutor implementation as an alternative.
Optional (defaults to SimpleAsyncTaskExecutor ). |
| 23 | By default, the underlying AbstractMessageListenerContainer creates a new instance of the DefaultTransactionAttribute (it takes the EJB approach to rolling back on runtime but not checked exceptions).
Optional (defaults to DefaultTransactionAttribute ). |
| 24 | Sets a bean reference to an external PlatformTransactionManager on the underlying AbstractMessageListenerContainer .
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 , the commit or rollback of the messaging transaction is deferred until the end of the current transaction.
If the channelTransacted flag is false , no transaction semantics apply to the messaging operation (it is auto-acked).
For further information, see
Transactions with Spring AMQP.
Optional. |
| 25 | Tells the SimpleMessageListenerContainer how many messages to process in a single transaction (if the channel is transactional).
For best results, it should be less than or equal to the value set in prefetch-count .
Not allowed when 'consumers-per-queue' is set.
Optional (defaults to 1 ). |
| 26 | Indicates that the underlying listener container should be a DirectMessageListenerContainer instead of the default SimpleMessageListenerContainer .
See the Spring AMQP Reference Manual for more information. |
|
container
Note that when configuring an external container, you cannot use the Spring AMQP namespace to define the container.
This is because the namespace requires at least one
|
Even though the Spring Integration JMS and AMQP support is similar, important differences exist.
The JMS inbound channel adapter is using a JmsDestinationPollingSource under the covers and expects a configured poller.
The AMQP inbound channel adapter uses an AbstractMessageListenerContainer and is message driven.
In that regard, it is more similar to the JMS message-driven channel adapter.
|
14.1.1. Configuring with Java Configuration
The following Spring Boot application shows an example of configuring the inbound adapter with Java configuration:
@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("aName");
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());
}
};
}
}
14.1.2. Configuring with the Java DSL
The following Spring Boot application provides an example of configuring the inbound adapter with the Java DSL:
@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, "aName"))
.handle(m -> System.out.println(m.getPayload()))
.get();
}
}
14.2. Polled Inbound Channel Adapter
Version 5.0.1 introduced a polled channel adapter, letting you fetch individual messages on demand — for example, with a MessageSourcePollingTemplate
or a poller.
See Deferred Acknowledgment Pollable Message Source for more information.
It does not currently support XML configuration.
The following example shows how to configure an AmqpMessageSource
with Java configuration:
@Bean
public AmqpMessageSource source(ConnectionFactory connectionFactory) {
return new AmpqpMessageSource(connectionFactory, "someQueue");
}
See the Javadoc for configuration properties.
The following example shows how to configure an inboundPolledAdapter
with the Java DSL:
@Bean
public IntegrationFlow flow() {
return IntegrationFlows.from(Amqp.inboundPolledAdapter(connectionFactory(), DSL_QUEUE),
e -> e.poller(Pollers.fixedDelay(1_000)).autoStartup(false))
.handle(p -> {
...
})
.get();
}
14.3. Inbound Gateway
The inbound gateway supports all the attributes on the inbound channel adapter (except that 'channel' is replaced by 'request-channel'), plus some additional attributes. The following listing shows the available attributes:
<int-amqp:inbound-gateway
id="inboundGateway" (1)
request-channel="myRequestChannel" (2)
header-mapper="" (3)
mapped-request-headers="" (4)
mapped-reply-headers="" (5)
reply-channel="myReplyChannel" (6)
reply-timeout="1000" (7)
amqp-template="" (8)
default-reply-to="" /> (9)
| 1 | The Unique ID for this adapter. Optional. |
| 2 | Message channel to which converted messages are sent. Required. |
| 3 | A reference to an AmqpHeaderMapper to use when receiving AMQP Messages.
Optional.
By default, only standard AMQP properties (such as contentType ) are copied to and from Spring Integration MessageHeaders .
Any user-defined headers within the AMQP MessageProperties are not copied to or from an AMQP message by the default DefaultAmqpHeaderMapper .
Not allowed if 'request-header-names' or 'reply-header-names' is provided. |
| 4 | Comma-separated list of names of AMQP Headers to be mapped from the AMQP request into the MessageHeaders .
This attribute can be provided only if the 'header-mapper' reference is not provided.
The values in this list can also be simple patterns to be matched against the header names (e.g. "*" or "thing1*, thing2" or "*thing1" ). |
| 5 | Comma-separated list of names of MessageHeaders to be mapped into the AMQP message properties of the AMQP reply message.
All standard Headers (such as contentType ) are mapped to AMQP Message Properties, while user-defined headers are mapped to the 'headers' property.
This attribute can only be provided if the 'header-mapper' reference is not provided.
The values in this list can also be simple patterns to be matched against the header names (for example, "*" or "foo*, bar" or "*foo" ). |
| 6 | Message Channel where reply Messages are expected. Optional. |
| 7 | Sets the receiveTimeout on the underlying o.s.i.core.MessagingTemplate for receiving messages from the reply channel.
If not specified, this property defaults to 1000 (1 second).
Only applies if the container thread hands off to another thread before the reply is sent. |
| 8 | The customized AmqpTemplate bean reference (to have more control over the reply messages to send).
You can provide an alternative implementation to the RabbitTemplate . |
| 9 | The replyTo o.s.amqp.core.Address to be used when the requestMessage does not have a replyTo
property.
If this option is not specified, no amqp-template is provided, no replyTo property exists in the request message, and
an IllegalStateException is thrown because the reply cannot be routed.
If this option is not specified and an external amqp-template is provided, no exception is thrown.
You must either specify this option or configure a default exchange and routingKey on that template,
if you anticipate cases when no replyTo property exists in the request message. |
See the note in Inbound Channel Adapter about configuring the listener-container
attribute.
14.3.1. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the inbound gateway with Java configuration:
@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();
}
};
}
}
14.3.2. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the inbound gateway with the Java DSL:
@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();
}
}
14.4. Inbound Endpoint Acknowledge Mode
By default, the inbound endpoints use the AUTO
acknowledge mode, which means the container automatically acknowledges the message when the downstream integration flow completes (or a message is handed off to another thread by using a QueueChannel
or ExecutorChannel
).
Setting the mode to NONE
configures the consumer such that acknowledgments are not used at all (the broker automatically acknowledges the message as soon as it is sent).
Setting the mode to MANUAL
lets user code acknowledge the message at some other point during processing.
To support this, with this mode, the endpoints provide the Channel
and deliveryTag
in the amqp_channel
and amqp_deliveryTag
headers, respectively.
You can perform any valid Rabbit command on the Channel
but, generally, only basicAck
and basicNack
(or basicReject
) are used.
In order to not interfere with the operation of the container, you should not retain a reference to the channel and use it only in the context of the current message.
Since the Channel is a reference to a “live” object, it cannot be serialized and is lost if a message is persisted.
|
The following example shows how you might use MANUAL
acknowledgement:
@ServiceActivator(inputChannel = "foo", outputChannel = "bar")
public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Channel channel,
@Header(AmqpHeaders.DELIVERY_TAG) Long deliveryTag) throws Exception {
// Do some processing
if (allOK) {
channel.basicAck(deliveryTag, false);
// perhaps do some more processing
}
else {
channel.basicNack(deliveryTag, false, true);
}
return someResultForDownStreamProcessing;
}
14.5. Outbound Channel Adapter
The following example shows the available properties for an AMQP outbound channel adapter:
<int-amqp:outbound-channel-adapter id="outboundAmqp" (1)
channel="outboundChannel" (2)
amqp-template="myAmqpTemplate" (3)
exchange-name="" (4)
exchange-name-expression="" (5)
order="1" (6)
routing-key="" (7)
routing-key-expression="" (8)
default-delivery-mode"" (9)
confirm-correlation-expression="" (10)
confirm-ack-channel="" (11)
confirm-nack-channel="" (12)
return-channel="" (13)
error-message-strategy="" (14)
header-mapper="" (15)
mapped-request-headers="" (16)
lazy-connect="true" /> (17)
| 1 | The unique ID for this adapter. Optional. |
| 2 | Message channel to which messages should be sent to have them converted and published to an AMQP exchange. Required. |
| 3 | Bean reference to the configured AMQP template.
Optional (defaults to amqpTemplate ). |
| 4 | The name of the AMQP exchange to which messages are sent. If not provided, messages are sent to the default, no-name exchange. Mutually exclusive with 'exchange-name-expression'. Optional. |
| 5 | A SpEL expression that is evaluated to determine the name of the AMQP exchange to which messages are sent, with the message as the root object. If not provided, messages are sent to the default, no-name exchange. Mutually exclusive with 'exchange-name'. Optional. |
| 6 | The order for this consumer when multiple consumers are registered, thereby enabling load-balancing and failover.
Optional (defaults to Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE] ). |
| 7 | The fixed routing-key to use when sending messages.
By default, this is an empty String .
Mutually exclusive with 'routing-key-expression'.
Optional. |
| 8 | A SpEL expression that is evaluated to determine the routing key to use when sending messages, with the message as the root object (for example, 'payload.key').
By default, this is an empty String .
Mutually exclusive with 'routing-key'.
Optional. |
| 9 | The default delivery mode for messages: PERSISTENT or NON_PERSISTENT .
Overridden if the header-mapper sets the delivery mode.
If the Spring Integration message header amqp_deliveryMode is present, the DefaultHeaderMapper sets the value.
If this attribute is not supplied and the header mapper does not 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. |
| 10 | An expression that defines correlation data.
When provided, this configures the underlying AMQP template to receive publisher confirmations.
Requires a dedicated RabbitTemplate and a CachingConnectionFactory with the publisherConfirms property set to true .
When a publisher confirmation is received and correlation data is supplied, it is written to either the confirm-ack-channel or the confirm-nack-channel , depending on the confirmation type.
The payload of the confirmation is the correlation data, as defined by this expression.
The message has an 'amqp_publishConfirm' header set to true (ack ) or false (nack ).
Examples: headers['myCorrelationData'] and payload .
Version 4.1 introduced the amqp_publishConfirmNackCause message header.
It contains the cause of a 'nack' for a publisher confirmation.
Starting with version 4.2, if the expression resolves to a Message<?> instance (such as #this ), the message emitted on the ack /nack channel is based on that message, with the additional header(s) added.
Previously, a new message was created with the correlation data as its payload, regardless of type.
Optional. |
| 11 | The channel to which positive (ack ) publisher confirms are sent.
The payload is the correlation data defined by the confirm-correlation-expression .
If the expression is #root or #this , the message is built from the original message, with the amqp_publishConfirm header set to true .
Optional (the default is nullChannel ). |
| 12 | The channel to which negative (nack ) publisher confirmations are sent.
The payload is the correlation data defined by the confirm-correlation-expression (if there is no ErrorMessageStrategy configured).
If the expression is #root or #this , the message is built from the original message, with the amqp_publishConfirm header set to false .
When there is an ErrorMessageStrategy , the message is an ErrorMessage with a NackedAmqpMessageException payload.
Optional (the default is nullChannel ). |
| 13 | The channel to which returned messages are sent.
When provided, the underlying AMQP template is configured to return undeliverable messages to the adapter.
When there is no ErrorMessageStrategy configured, the message is constructed from the data received from AMQP, with the following additional headers: amqp_returnReplyCode , amqp_returnReplyText , amqp_returnExchange , amqp_returnRoutingKey .
When there is an ErrorMessageStrategy , the message is an ErrorMessage with a ReturnedAmqpMessageException payload.
Optional. |
| 14 | A reference to an ErrorMessageStrategy implementation used to build ErrorMessage instances when sending returned or negatively acknowledged messages. |
| 15 | A reference to an AmqpHeaderMapper to use when sending AMQP Messages.
By default, only standard AMQP properties (such as contentType ) are copied to the Spring Integration MessageHeaders .
Any user-defined headers is not copied to the message by the default`DefaultAmqpHeaderMapper`.
Not allowed if 'request-header-names' is provided.
Optional. |
| 16 | Comma-separated list of names of AMQP Headers to be mapped from the MessageHeaders to the AMQP Message.
Not allowed if the 'header-mapper' reference is provided.
The values in this list can also be simple patterns to be matched against the header names (e.g. "*" or "thing1*, thing2" or "*thing1" ). |
| 17 | When set to false , the endpoint attempts to connect to the broker during application context initialization.
This allows “fail fast” detection of bad configuration but also causes initialization to fail if the broker is down.
When true (the default), the connection is established (if it does not already exist because some other component established it) when the first message is sent. |
|
return-channel
Using a |
14.5.1. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the outbound adapter with Java configuration:
@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);
}
}
14.5.2. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the outbound adapter with the Java DSL:
@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);
}
}
14.6. Outbound Gateway
The following listing shows the possible properties for an AMQP Outbound Gateway:
<int-amqp:outbound-gateway id="inboundGateway" (1)
request-channel="myRequestChannel" (2)
amqp-template="" (3)
exchange-name="" (4)
exchange-name-expression="" (5)
order="1" (6)
reply-channel="" (7)
reply-timeout="" (8)
requires-reply="" (9)
routing-key="" (10)
routing-key-expression="" (11)
default-delivery-mode"" (12)
confirm-correlation-expression="" (13)
confirm-ack-channel="" (14)
confirm-nack-channel="" (15)
return-channel="" (16)
error-message-strategy="" (17)
lazy-connect="true" /> (18)
| 1 | The unique ID for this adapter. Optional. |
| 2 | Message channel to which messages are sent to have them converted and published to an AMQP exchange. Required. |
| 3 | Bean reference to the configured AMQP template.
Optional (defaults to amqpTemplate ). |
| 4 | The name of the AMQP exchange to which messages should be sent. If not provided, messages are sent to the default, no-name cxchange. Mutually exclusive with 'exchange-name-expression'. Optional. |
| 5 | A SpEL expression that is evaluated to determine the name of the AMQP exchange to which messages should be sent, with the message as the root object. If not provided, messages are sent to the default, no-name exchange. Mutually exclusive with 'exchange-name'. Optional. |
| 6 | The order for this consumer when multiple consumers are registered, thereby enabling load-balancing and failover.
Optional (defaults to Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE] ). |
| 7 | Message channel to which replies should be sent after being received from an AMQP queue and converted. Optional. |
| 8 | The time the gateway waits when sending the reply message to the reply-channel .
This only applies if the reply-channel can block — such as a QueueChannel with a capacity limit that is currently full.
Defaults to infinity. |
| 9 | When true , the gateway throws an exception if no reply message is received within the AmqpTemplate’s `replyTimeout property.
Defaults to true . |
| 10 | The routing-key to use when sending messages.
By default, this is an empty String .
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 (for example, 'payload.key').
By default, this is an empty String .
Mutually exclusive with 'routing-key'.
Optional. |
| 12 | The default delivery mode for messages: PERSISTENT or NON_PERSISTENT .
Overridden if the header-mapper sets the delivery mode.
If the Spring Integration message header amqp_deliveryMode is present, the DefaultHeaderMapper sets the value.
If this attribute is not supplied and the header mapper does not 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. |
| 13 | Since version 4.2.
An expression defining correlation data.
When provided, this configures the underlying AMQP template to receive publisher confirms.
Requires a dedicated RabbitTemplate and a CachingConnectionFactory with the publisherConfirms property set to true .
When a publisher confirm is received and correlation data is supplied, it is written to either the confirm-ack-channel or the confirm-nack-channel , depending on the confirmation type.
The payload of the confirm is the correlation data, as defined by this expression.
The message has a header 'amqp_publishConfirm' set to true (ack ) or false (nack ).
For nack confirmations, Spring Integration provides an additional header amqp_publishConfirmNackCause .
Examples: headers['myCorrelationData'] and payload .
If the expression resolves to a Message<?> instance (such as #this ), the message
emitted on the ack /nack channel is based on that message, with the additional headers added.
Previously, a new message was created with the correlation data as its payload, regardless of type.
Optional. |
| 14 | The channel to which positive (ack ) publisher confirmations are sent.
The payload is the correlation data defined by confirm-correlation-expression .
If the expression is #root or #this , the message is built from the original message, with the amqp_publishConfirm header set to true .
Optional (the default is nullChannel ). |
| 15 | The channel to which negative (nack ) publisher confirmations are sent.
The payload is the correlation data defined by confirm-correlation-expression (if there is no ErrorMessageStrategy configured).
If the expression is #root or #this , the message is built from the original message, with the amqp_publishConfirm header set to false .
When there is an ErrorMessageStrategy , the message is an ErrorMessage with a NackedAmqpMessageException payload.
Optional (the default is nullChannel ). |
| 16 | The channel to which returned messages are sent.
When provided, the underlying AMQP template is configured to return undeliverable messages to the adapter.
When there is no ErrorMessageStrategy configured, the message is constructed from the data received from AMQP, with the following additional headers: amqp_returnReplyCode , amqp_returnReplyText , amqp_returnExchange , and amqp_returnRoutingKey .
When there is an ErrorMessageStrategy , the message is an ErrorMessage with a ReturnedAmqpMessageException payload.
Optional. |
| 17 | A reference to an ErrorMessageStrategy implementation used to build ErrorMessage instances when sending returned or negatively acknowledged messages. |
| 18 | When set to false , the endpoint attempts to connect to the broker during application context initialization.
This allows “fail fast” detection of bad configuration by logging an error message if the broker is down.
When true (the default), the connection is established (if it does not already exist because some other component established it) when the first message is sent. |
|
return-channel
Using a |
The underlying AmqpTemplate has a default replyTimeout of five seconds.
If you require a longer timeout, you must configure it on the template .
|
14.6.1. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the outbound gateway with Java configuration:
@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);
}
}
Note that the only difference between the outbound adapter and outbound gateway configuration is the setting of the
expectReply
property.
14.6.2. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the outbound adapter with the Java DSL:
@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);
}
}
14.7. Asynchronous Outbound Gateway
The gateway discussed in the previous section is synchronous, in that the sending thread is suspended until a
reply is received (or a timeout occurs).
Spring Integration version 4.3 added an asynchronous gateway, which uses the AsyncRabbitTemplate
from Spring AMQP.
When a message is sent, the thread returns immediately after the send operation completes, and, when the message is received, the reply is sent on the template’s listener container thread.
This can be useful when the gateway is invoked on a poller thread.
The thread is released and is available for other tasks in the framework.
The following listing shows the possible configuration options for an AMQP asynchronous outbound gateway:
<int-amqp:outbound-gateway id="inboundGateway" (1)
request-channel="myRequestChannel" (2)
async-template="" (3)
exchange-name="" (4)
exchange-name-expression="" (5)
order="1" (6)
reply-channel="" (7)
reply-timeout="" (8)
requires-reply="" (9)
routing-key="" (10)
routing-key-expression="" (11)
default-delivery-mode"" (12)
confirm-correlation-expression="" (13)
confirm-ack-channel="" (14)
confirm-nack-channel="" (15)
return-channel="" (16)
lazy-connect="true" /> (17)
| 1 | The unique ID for this adapter. Optional. |
| 2 | Message channel to which messages should be sent in order to have them converted and published to an AMQP exchange. Required. |
| 3 | Bean reference to the configured AsyncRabbitTemplate .
Optional (it defaults to asyncRabbitTemplate ). |
| 4 | The name of the AMQP exchange to which messages should be sent. If not provided, messages are sent to the default, no-name exchange. Mutually exclusive with 'exchange-name-expression'. Optional. |
| 5 | A SpEL expression that is evaluated to determine the name of the AMQP exchange to which messages are sent, with the message as the root object. If not provided, messages are sent to the default, no-name exchange. Mutually exclusive with 'exchange-name'. Optional. |
| 6 | The order for this consumer when multiple consumers are registered, thereby enabling load-balancing and failover.
Optional (it defaults to Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE] ). |
| 7 | Message channel to which replies should be sent after being received from an AMQP queue and converted. Optional. |
| 8 | The time the gateway waits when sending the reply message to the reply-channel .
This only applies if the reply-channel can block — such as a QueueChannel with a capacity limit that is currently full.
The default is infinity. |
| 9 | When no reply message is received within the AsyncRabbitTemplate’s `receiveTimeout property and this setting is true , the gateway sends an error message to the inbound message’s errorChannel header.
When no reply message is received within the AsyncRabbitTemplate’s `receiveTimeout property and this setting is false , the gateway sends an error message to the default errorChannel (if available).
It defaults to true . |
| 10 | The routing-key to use when sending Messages.
By default, this is an empty String .
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 (for example, 'payload.key').
By default, this is an empty String .
Mutually exclusive with 'routing-key'.
Optional. |
| 12 | The default delivery mode for messages: PERSISTENT or NON_PERSISTENT .
Overridden if the header-mapper sets the delivery mode.
If the Spring Integration message header (amqp_deliveryMode ) is present, the DefaultHeaderMapper sets the value.
If this attribute is not supplied and the header mapper does not set it, the default depends on the underlying Spring AMQP MessagePropertiesConverter used by the RabbitTemplate .
If that is not customized, the default is PERSISTENT .
Optional. |
| 13 | An expression that defines correlation data.
When provided, this configures the underlying AMQP template to receive publisher confirmations.
Requires a dedicated RabbitTemplate and a CachingConnectionFactory with its publisherConfirms property set to true .
When a publisher confirmation is received and correlation data is supplied, the confirmation is written to either the confirm-ack-channel or the confirm-nack-channel , depending on the confirmation type.
The payload of the confirmation is the correlation data as defined by this expression, and the message has its 'amqp_publishConfirm' header set to true (ack ) or false (nack ).
For nack instances, an additional header (amqp_publishConfirmNackCause ) is provided.
Examples: headers['myCorrelationData'] , payload .
If the expression resolves to a Message<?> instance (such as “#this”), the message emitted on the ack /nack channel is based on that message, with the additional headers added.
Optional. |
| 14 | The channel to which positive (ack ) publisher confirmations are sent.
The payload is the correlation data defined by the confirm-correlation-expression .
Requires the underlying AsyncRabbitTemplate to have its enableConfirms property set to true .
Optional (the default is nullChannel ). |
| 15 | Since version 4.2.
The channel to which negative (nack ) publisher confirmations are sent.
The payload is the correlation data defined by the confirm-correlation-expression .
Requires the underlying AsyncRabbitTemplate to have its enableConfirms property set to true .
Optional (the default is nullChannel ). |
| 16 | The channel to which returned messages are sent.
When provided, the underlying AMQP template is configured to return undeliverable messages to the gateway.
The message is constructed from the data received from AMQP, with the following additional headers: amqp_returnReplyCode , amqp_returnReplyText , amqp_returnExchange , and amqp_returnRoutingKey .
Requires the underlying AsyncRabbitTemplate to have its mandatory property set to true .
Optional. |
| 17 | When set to false , the endpoint tries to connect to the broker during application context initialization.
Doing so allows “fail fast” detection of bad configuration, by logging an error message if the broker is down.
When true (the default), the connection is established (if it does not already exist because some other component established
it) when the first message is sent. |
See also Asynchronous Service Activator for more information.
|
RabbitTemplate
When you use confirmations and returns, we recommend that the |
14.7.1. Configuring with Java Configuration
The following configuration shows an example of how to configure the outbound gateway with Java configuration:
@Configuration
public class AmqpAsyncConfig {
@Bean
@ServiceActivator(inputChannel = "amqpOutboundChannel")
public AsyncAmqpOutboundGateway amqpOutbound(AmqpTemplate asyncTemplate) {
AsyncAmqpOutboundGateway outbound = new AsyncAmqpOutboundGateway(asyncTemplate);
outbound.setRoutingKey("foo"); // default exchange - route to queue 'foo'
return outbound;
}
@Bean
public AsyncRabbitTemplate asyncTemplate(RabbitTemplate rabbitTemplate,
SimpleMessageListenerContainer replyContainer) {
return new AsyncRabbitTemplate(rabbitTemplate, replyContainer);
}
@Bean
public SimpleMessageListenerContainer replyContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(ccf);
container.setQueueNames("asyncRQ1");
return container;
}
@Bean
public MessageChannel amqpOutboundChannel() {
return new DirectChannel();
}
}
14.7.2. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the outbound adapter with the Java DSL:
@SpringBootApplication
public class AmqpAsyncApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(AmqpAsyncApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
String reply = gateway.sendToRabbit("foo");
System.out.println(reply);
}
@Bean
public IntegrationFlow asyncAmqpOutbound(AsyncRabbitTemplate asyncRabbitTemplate) {
return f -> f
.handle(Amqp.asyncOutboundGateway(asyncRabbitTemplate)
.routingKey("foo")); // default exchange - route to queue 'foo'
}
@MessagingGateway(defaultRequestChannel = "asyncAmqpOutbound.input")
public interface MyGateway {
String sendToRabbit(String data);
}
}
14.8. Inbound Message Conversion
Inbound messages, arriving at the channel adapter or gateway, are converted to the spring-messaging
Message<?>
payload using a message converter.
By default, a SimpleMessageConverter
is used, which handles java serialization and text.
Headers are mapped using the DefaultHeaderMapper.inboundMapper()
by default.
If a conversion error occurs, and there is no error channel defined, the exception is thrown to the container and handled by the listener container’s error handler.
The default error handler treats conversion errors as fatal and the message will be rejected (and routed to a dead-letter exchange, if the queue is so configured).
If an error channel is defined, the ErrorMessage
payload is a ListenerExecutionFailedException
with properties failedMessage
(the Spring AMQP message that could not be converted) and the cause
.
If the container AcknowledgeMode
is AUTO
(the default) and the error flow consumes the error without throwing an exception, the original message will be acknowledged.
If the error flow throws an exception, the exception type, in conjunction with the container’s error handler, will determine whether or not the message is requeued.
If the container is configured with AcknowledgeMode.MANUAL
, the payload is a ManualAckListenerExecutionFailedException
with additional properties channel
and deliveryTag
.
This enables the error flow to call basicAck
or basicNack
(or basicReject
) for the message, to control its disposition.
14.9. Outbound Message Conversion
Spring AMQP 1.4 introduced the ContentTypeDelegatingMessageConverter
, where the actual converter is selected based
on the incoming content type message property.
This can be used by inbound endpoints.
As of Spring Integration version 4.3, you can use the ContentTypeDelegatingMessageConverter
on outbound endpoints as well, with the contentType
header specifying which converter is used.
The following example configures a ContentTypeDelegatingMessageConverter
, with the default converter being the SimpleMessageConverter
(which handles Java serialization and plain text), together with a JSON converter:
<amqp:outbound-channel-adapter id="withContentTypeConverter" channel="ctRequestChannel"
exchange-name="someExchange"
routing-key="someKey"
amqp-template="amqpTemplateContentTypeConverter" />
<int:channel id="ctRequestChannel"/>
<rabbit:template id="amqpTemplateContentTypeConverter"
connection-factory="connectionFactory" message-converter="ctConverter" />
<bean id="ctConverter"
class="o.s.amqp.support.converter.ContentTypeDelegatingMessageConverter">
<property name="delegates">
<map>
<entry key="application/json">
<bean class="o.s.amqp.support.converter.Jackson2JsonMessageConverter" />
</entry>
</map>
</property>
</bean>
Sending a message to ctRequestChannel
with the contentType
header set to application/json
causes the JSON converter to be selected.
This applies to both the outbound channel adapter and gateway.
|
Starting with version 5.0, headers that are added to the There are, however, cases where the previous behavior is desired — for example, when a There is now a property called |
14.10. Outbound User ID