diff --git a/src/reference/asciidoc/barrier.adoc b/src/reference/asciidoc/barrier.adoc
index 42cde75d84..cfc236347f 100644
--- a/src/reference/asciidoc/barrier.adoc
+++ b/src/reference/asciidoc/barrier.adoc
@@ -35,7 +35,27 @@ An exception is thrown if a second thread arrives with the same correlation.
The following example shows how to use a custom header for correlation:
-[source, xml]
+====
+[source, java, role="primary"]
+.Java
+----
+@ServiceActivator(inputChannel="in")
+@Bean
+public BarrierMessageHandler barrier(MessageChannel out, MessageChannel lateTriggerChannel) {
+ BarrierMessageHandler barrier = new BarrierMessageHandler(10000);
+ barrier.setOutputChannel(out());
+ barrier.setDiscardChannel(lateTriggerChannel);
+ return barrier;
+}
+
+@ServiceActivator (inputChannel="release")
+@Bean
+public MessageHandler releaser(MessageTriggerAction barrier) {
+ return barrier::trigger(message);
+}
+----
+[source, xml, role="secondary"]
+.XML
----
----
+====
Depending on which one has a message arrive first, either the thread sending a message to `in` or the thread sending a message to `release` waits for up to ten seconds until the other message arrives.
When the message is released, the `out` channel is sent a message that combines the result of invoking the custom `MessageGroupProcessor` bean, named `myOutputProcessor`.
If the main thread times out and a trigger arrives later, you can configure a discard channel to which the late trigger is sent.
-The following example shows the Java configuration to do so:
-
-[source, java]
-----
-@Configuration
-@EnableIntegration
-public class Config {
-
- @ServiceActivator(inputChannel="in")
- @Bean
- public BarrierMessageHandler barrier() {
- BarrierMessageHandler barrier = new BarrierMessageHandler(10000);
- barrier.setOutputChannel(out());
- barrier.setDiscardChannel(lateTriggers());
- return barrier;
- }
-
- @ServiceActivator (inputChannel="release")
- @Bean
- public MessageHandler releaser() {
- return new MessageHandler() {
-
- @Override
- public void handleMessage(Message message) throws MessagingException {
- barrier().trigger(message);
- }
-
- };
- }
-
-}
-----
For an example of this component, see the https://github.com/spring-projects/spring-integration-samples/tree/main/basic/barrier[barrier sample application].
diff --git a/src/reference/asciidoc/chain.adoc b/src/reference/asciidoc/chain.adoc
index b56ff93e4c..07238c43e2 100644
--- a/src/reference/asciidoc/chain.adoc
+++ b/src/reference/asciidoc/chain.adoc
@@ -7,6 +7,10 @@ For example, it is fairly common to provide a transformer before other component
Similarly, when you provide a filter before some other component in a chain, you essentially create a https://www.enterpriseintegrationpatterns.com/MessageSelector.html[selective consumer].
In either case, the chain requires only a single `input-channel` and a single `output-channel`, eliminating the need to define channels for each individual component.
+NOTE: The `MessageHandlerChain` is mostly designed for an XML configuration.
+For Java DSL, an `IntegrationFlow` definition can be treated as a chain component, but it has nothing to do with concepts and principles described in this chapter below.
+See <<./dsl.adoc#java-dsl,Java DSL>> for more information.
+
TIP: Spring Integration's `Filter` provides a boolean property: `throwExceptionOnRejection`.
When you provide multiple selective consumers on the same point-to-point channel with different acceptance criteria, you should set this value 'true' (the default is `false`) so that the dispatcher knows that the message was rejected and, as a result, tries to pass the message on to other subscribers.
If the exception were not thrown, it would appear to the dispatcher that the message had been passed on successfully even though the filter had dropped the message to prevent further processing.
diff --git a/src/reference/asciidoc/channel.adoc b/src/reference/asciidoc/channel.adoc
index d5d6ff5d46..d3449bfb01 100644
--- a/src/reference/asciidoc/channel.adoc
+++ b/src/reference/asciidoc/channel.adoc
@@ -360,53 +360,45 @@ NOTE: A less invasive approach that lets you invoke simple interfaces with paylo
[[channel-configuration]]
==== Configuring Message Channels
-To create a message channel instance, you can use the `` element, as follows:
+To create a message channel instance, you can use the `` element for xml or `DirectChannel` instance for Java configuration, as follows:
====
-[source,xml]
-----
-
-----
-====
-
-The equivalent Java configuration declares a `DirectChannel` `@Bean`:
-
-====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public MessageChannel exampleChannel() {
return new DirectChannel();
}
----
-====
-
-
-The default channel type is point-to-point.
-To create a publish-subscribe channel, use the `` element, as follows:
-
-====
-[source,xml]
+[source, xml, role="secondary"]
+.XML
----
-
+
----
====
-The Java configuration is:
+When you use the `` element without any sub-elements, it creates a `DirectChannel` instance (a `SubscribableChannel`).
+
+To create a publish-subscribe channel, use the `` element (the `PublishSubscribeChannel` in Java), as follows:
====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public MessageChannel exampleChannel() {
return new PublishSubscribeChannel();
}
----
+[source, xml, role="secondary"]
+.XML
+----
+
+----
====
-When you use the `` element without any sub-elements, it creates a `DirectChannel` instance (a `SubscribableChannel`).
-
-However, you can alternatively provide a variety of `` sub-elements to create any of the pollable channel types (as described in <>).
+You can alternatively provide a variety of `` sub-elements to create any of the pollable channel types (as described in <>).
The following sections shows examples of each channel type.
[[channel-configuration-directchannel]]
@@ -416,44 +408,27 @@ As mentioned earlier, `DirectChannel` is the default type.
The following listing shows who to define one:
====
-[source,xml]
-----
-
-----
-====
-
-In Java Configuration:
-
-====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public MessageChannel directChannel() {
return new DirectChannel();
}
----
+[source, xml, role="secondary"]
+.XML
+----
+
+----
====
A default channel has a round-robin load-balancer and also has failover enabled (see <> for more detail).
-To disable one or both of these, add a `` sub-element and configure the attributes as follows:
+To disable one or both of these, add a `` sub-element (a `LoadBalancingStrategy` constructor of the `DirectChannel`) and configure the attributes as follows:
====
-[source,xml]
-----
-
-
-
-
-
-
-
-----
-====
-
-In Java Configuration:
-
-====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public MessageChannel failFastChannel() {
@@ -467,6 +442,17 @@ public MessageChannel failFastChannel() {
return new DirectChannel(null);
}
----
+[source, xml, role="secondary"]
+.XML
+----
+
+
+
+
+
+
+
+----
====
[[channel-datatype-channel]]
@@ -482,16 +468,8 @@ You can use separate datatype channels for each specific payload data type.
To create a datatype channel that accepts only messages that contain a certain payload type, provide the data type's fully-qualified class name in the channel element's `datatype` attribute, as the following example shows:
====
-[source,xml]
-----
-
-----
-====
-
-In Java Configuration:
-
-====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public MessageChannel numberChannel() {
@@ -500,6 +478,11 @@ public MessageChannel numberChannel() {
return channel;
}
----
+[source, xml, role="secondary"]
+.XML
+----
+
+----
====
Note that the type check passes for any type that is assignable to the channel's datatype.
@@ -507,7 +490,18 @@ In other words, the `numberChannel` in the preceding example would accept messag
Multiple types can be provided as a comma-delimited list, as the following example shows:
====
-[source,xml]
+[source, java, role="primary"]
+.Java
+----
+@Bean
+public MessageChannel numberChannel() {
+ DirectChannel channel = new DirectChannel();
+ channel.setDatatypes(String.class, Number.class);
+ return channel;
+}
+----
+[source, xml, role="secondary"]
+.XML
----
----
@@ -563,18 +557,8 @@ public static class StringToIntegerConverter implements Converter
-
-
-----
-====
-
-With Java Configuration you must use an `@IntegrationConverter` next to a `@Bean` annotation:
-
-====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
@IntegrationConverter
@@ -582,6 +566,13 @@ public StringToIntegerConverter strToInt {
return new StringToIntegerConverter();
}
----
+[source, xml, role="secondary"]
+.XML
+----
+
+
+
+----
====
Or on the `StringToIntegerConverter` class when it is marked with the `@Component` annotation for auto-scanning.
@@ -608,7 +599,16 @@ To create a `QueueChannel`, use the `` sub-element.
You may specify the channel's capacity as follows:
====
-[source,xml]
+[source, java, role="primary"]
+.Java
+----
+@Bean
+public PollableChannel queueChannel() {
+ return new QueueChannel(25);
+}
+----
+[source, xml, role="secondary"]
+.XML
----
@@ -616,22 +616,9 @@ You may specify the channel's capacity as follows:
----
====
-
NOTE: If you do not provide a value for the 'capacity' attribute on this `` sub-element, the resulting queue is unbounded.
To avoid issues such as running out of memory, we highly recommend that you set an explicit value for a bounded queue.
-With Java Configuration:
-
-====
-[source,java]
-----
-@Bean
-public PollableChannel queueChannel() {
- return new QueueChannel(25);
-}
-----
-====
-
====== Persistent `QueueChannel` Configuration
Since a `QueueChannel` provides the capability to buffer messages but does so in-memory only by default, it also introduces a possibility that messages could be lost in the event of a system failure.
@@ -662,7 +649,7 @@ You can configure a message store for any `QueueChannel` by adding the `message-
----
====
-(See samples below for Java Configuration options.)
+(See samples below for Java/Kotlin Configuration options.)
The Spring Integration JDBC module also provides a schema Data Definition Language (DDL) for a number of popular databases.
These schemas are located in the org.springframework.integration.jdbc.store.channel package of that module (`spring-integration-jdbc`).
@@ -680,7 +667,8 @@ The notion of priority is determined by the message store implementation.
For example, the following example shows the Java configuration for the <<./mongodb.adoc#mongodb-priority-channel-message-store,MongoDB Channel Message Store>>:
====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public BasicMessageGroupStore mongoDbChannelMessageStore(MongoDbFactory mongoDbFactory) {
@@ -694,14 +682,8 @@ public PollableChannel priorityQueue(BasicMessageGroupStore mongoDbChannelMessag
return new PriorityChannel(new MessageGroupQueue(mongoDbChannelMessageStore, "priorityQueue"));
}
----
-====
-
-NOTE: Pay attention to the `MessageGroupQueue` class.
-That is a `BlockingQueue` implementation to use the `MessageGroupStore` operations.
-
-The same implementation with Java DSL might look like the following example:
-
-[source,java]
+[source, java, role="secondary"]
+.Java DSL
----
@Bean
public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) {
@@ -711,6 +693,19 @@ public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbCh
.get();
}
----
+[source, kotlin, role="secondary"]
+.Kotlin DSL
+----
+@Bean
+fun priorityFlow(mongoDbChannelMessageStore: PriorityCapableChannelMessageStore) =
+ integrationFlow {
+ channel { priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup") }
+ }
+----
+====
+
+NOTE: Pay attention to the `MessageGroupQueue` class.
+That is a `BlockingQueue` implementation to use the `MessageGroupStore` operations.
Another option to customize the `QueueChannel` environment is provided by the `ref` attribute of the `` sub-element or its particular constructor.
This attribute supplies the reference to any `java.util.Queue` implementation.
@@ -740,22 +735,19 @@ To create a `PublishSubscribeChannel`, use the elem
When using this element, you can also specify the `task-executor` used for publishing messages (if none is specified, it publishes in the sender's thread), as follows:
====
-[source,xml]
-----
-
-----
-====
-
-With Java Configuration:
-
-====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public MessageChannel pubsubChannel() {
return new PublishSubscribeChannel(someExecutor());
}
----
+[source, xml, role="secondary"]
+.XML
+----
+
+----
====
If you provide a resequencer or aggregator downstream from a `PublishSubscribeChannel`, you can set the 'apply-sequence' property on the channel to `true`.
@@ -773,14 +765,8 @@ For example, if there are five subscribers, the sequence-size would be set to `5
The following example shows how to set the `apply-sequence` header to `true`:
====
-[source,xml]
-----
-
-----
-====
-
-====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public MessageChannel pubsubChannel() {
@@ -789,6 +775,11 @@ public MessageChannel pubsubChannel() {
return channel;
}
----
+[source, xml, role="secondary"]
+.XML
+----
+
+----
====
NOTE: The `apply-sequence` value is `false` by default so that a publish-subscribe channel can send the exact same message instances to multiple outbound channels.
@@ -807,24 +798,21 @@ As mentioned earlier, doing so breaks the single-threaded execution context betw
The following example shows how to use the `dispatcher` element and specify an executor in the `task-executor` attribute:
====
-[source,xml]
-----
-
-
-
-----
-====
-
-In Java Configuration you must use an `ExecutorChannel` bean definition:
-
-====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public MessageChannel executorChannel() {
return new ExecutorChannel(someExecutor());
}
----
+[source, xml, role="secondary"]
+.XML
+----
+
+
+
+----
====
[NOTE]
@@ -849,24 +837,21 @@ Consequently, the channel has a round-robin load-balancing strategy with failove
To create a `PriorityChannel`, use the `` sub-element, as the following example shows:
====
-[source,xml]
-----
-
-
-
-----
-====
-
-In JavaConfiguration:
-
-====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public PollableChannel priorityChannel() {
return new PriorityChannel(20);
}
----
+[source, xml, role="secondary"]
+.XML
+----
+
+
+
+----
====
By default, the channel consults the `priority` header of the message.
@@ -876,17 +861,8 @@ As with the `QueueChannel`, it also supports a `capacity` attribute.
The following example demonstrates all of these:
====
-[source,xml]
-----
-
-
-
-----
-====
-
-====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public PollableChannel priorityChannel() {
@@ -895,6 +871,14 @@ public PollableChannel priorityChannel() {
return channel;
}
----
+[source, xml, role="secondary"]
+.XML
+----
+
+
+
+----
====
Since version 4.0, the `priority-channel` child element supports the `message-store` option (`comparator` and `capacity` are not allowed in that case).
@@ -911,22 +895,21 @@ It does not provide any additional configuration options to those described earl
The following example shows how to declare a `RendezvousChannel`:
====
-[source,xml]
-----
-
-
-
-----
-====
-
-====
-[source,java]
+[source, java, role="primary"]
+.Java
----
@Bean
public PollableChannel rendezvousChannel() {
return new RendezvousChannel();
}
----
+[source, xml, role="secondary"]
+.XML
+----
+
+
+
+----
====
[[channel-configuration-threadlocalchannel]]
@@ -1129,7 +1112,17 @@ Now, in addition to the normal `wire-tap` namespace support, the `pattern` and `
The following example shows how to configure a global wire tap:
====
-[source,xml]
+[source, java, role="primary"]
+.Java
+----
+@Bean
+@GlobalChannelInterceptor(patterns = "input*,thing2*,thing1", order = 3)
+public WireTap wireTap(MessageChannel wiretapChannel) {
+ return new WireTap(wiretapChannel);
+}
+----
+[source, xml, role="secondary"]
+.XML
----
----