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

Code switcher in barrier and channel docs

Browse Source 
* Mention in the `chain.adoc` that a `<chain>` pattern
is mostly for an XML configuration
This commit is contained in:
Artem Bilan
2021-05-10 16:14:53 -04:00
parent 65963c9d56
commit 49f0398144
3 changed files with 182 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
src/reference/asciidoc/barrier.adoc
View File

@@ -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
----
<int:barrier id="barrier1" input-channel="in" output-channel="out"
correlation-strategy-expression="headers['myHeader']"
@@ -46,41 +66,10 @@ The following example shows how to use a custom header for correlation:
<int:outbound-channel-adapter channel="release" ref="barrier1.handler" method="trigger" />
----
====
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].

4
src/reference/asciidoc/chain.adoc
View File

@@ -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.

319
src/reference/asciidoc/channel.adoc
View File

@@ -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 `<channel/>` element, as follows:
To create a message channel instance, you can use the `<channel/>` element for xml or `DirectChannel` instance for Java configuration, as follows:
====
[source,xml]
----
<int:channel id="exampleChannel"/>
----
====
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 `<publish-subscribe-channel/>` element, as follows:
====
[source,xml]
[source, xml, role="secondary"]
.XML
----
<int:publish-subscribe-channel id="exampleChannel"/>
<int:channel id="exampleChannel"/>
----
====
The Java configuration is:
When you use the `<channel/>` element without any sub-elements, it creates a `DirectChannel` instance (a `SubscribableChannel`).
To create a publish-subscribe channel, use the `<publish-subscribe-channel/>` 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
----
<int:publish-subscribe-channel id="exampleChannel"/>
----
====
When you use the `<channel/>` element without any sub-elements, it creates a `DirectChannel` instance (a `SubscribableChannel`).
However, you can alternatively provide a variety of `<queue/>` sub-elements to create any of the pollable channel types (as described in <<channel-implementations>>).
You can alternatively provide a variety of `<queue/>` sub-elements to create any of the pollable channel types (as described in <<channel-implementations>>).
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]
----
<int:channel id="directChannel"/>
----
====
In Java Configuration:
====
[source,java]
[source, java, role="primary"]
.Java
----
@Bean
public MessageChannel directChannel() {
return new DirectChannel();
}
----
[source, xml, role="secondary"]
.XML
----
<int:channel id="directChannel"/>
----
====
A default channel has a round-robin load-balancer and also has failover enabled (see <<channel-implementations-directchannel>> for more detail).
To disable one or both of these, add a `<dispatcher/>` sub-element and configure the attributes as follows:
To disable one or both of these, add a `<dispatcher/>` sub-element (a `LoadBalancingStrategy` constructor of the `DirectChannel`) and configure the attributes as follows:
====
[source,xml]
----
<int:channel id="failFastChannel">
<int:dispatcher failover="false"/>
</channel>
<int:channel id="channelWithFixedOrderSequenceFailover">
<int:dispatcher load-balancer="none"/>
</int:channel>
----
====
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
----
<int:channel id="failFastChannel">
<int:dispatcher failover="false"/>
</channel>
<int:channel id="channelWithFixedOrderSequenceFailover">
<int:dispatcher load-balancer="none"/>
</int:channel>
----
====
[[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]
----
<int:channel id="numberChannel" datatype="java.lang.Number"/>
----
====
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
----
<int:channel id="numberChannel" datatype="java.lang.Number"/>
----
====
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
----
<int:channel id="stringOrNumberChannel" datatype="java.lang.String,java.lang.Number"/>
----
@@ -563,18 +557,8 @@ public static class StringToIntegerConverter implements Converter<String, Intege
Then we can register it as a converter with the Integration Conversion Service, as the following example shows:
====
[source,xml]
----
<int:converter ref="strToInt"/>
<bean id="strToInt" class="org.springframework.integration.util.Demo.StringToIntegerConverter"/>
----
====
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
----
<int:converter ref="strToInt"/>
<bean id="strToInt" class="org.springframework.integration.util.Demo.StringToIntegerConverter"/>
----
====
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 `<queue/>` 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
----
<int:channel id="queueChannel">
<queue capacity="25"/>
@@ -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 `<queue/>` 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 `<int:queue>` 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 <publish-subscribe-channel/> 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]
----
<int:publish-subscribe-channel id="pubsubChannel" task-executor="someExecutor"/>
----
====
With Java Configuration:
====
[source,java]
[source, java, role="primary"]
.Java
----
@Bean
public MessageChannel pubsubChannel() {
return new PublishSubscribeChannel(someExecutor());
}
----
[source, xml, role="secondary"]
.XML
----
<int:publish-subscribe-channel id="pubsubChannel" task-executor="someExecutor"/>
----
====
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]
----
<int:publish-subscribe-channel id="pubsubChannel" apply-sequence="true"/>
----
====
====
[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
----
<int:publish-subscribe-channel id="pubsubChannel" apply-sequence="true"/>
----
====
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]
----
<int:channel id="executorChannel">
<int:dispatcher task-executor="someExecutor"/>
</int:channel>
----
====
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
----
<int:channel id="executorChannel">
<int:dispatcher task-executor="someExecutor"/>
</int:channel>
----
====
[NOTE]
@@ -849,24 +837,21 @@ Consequently, the channel has a round-robin load-balancing strategy with failove
To create a `PriorityChannel`, use the `<priority-queue/>` sub-element, as the following example shows:
====
[source,xml]
----
<int:channel id="priorityChannel">
<int:priority-queue capacity="20"/>
</int:channel>
----
====
In JavaConfiguration:
====
[source,java]
[source, java, role="primary"]
.Java
----
@Bean
public PollableChannel priorityChannel() {
return new PriorityChannel(20);
}
----
[source, xml, role="secondary"]
.XML
----
<int:channel id="priorityChannel">
<int:priority-queue capacity="20"/>
</int:channel>
----
====
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]
----
<int:channel id="priorityChannel" datatype="example.Widget">
<int:priority-queue comparator="widgetComparator"
capacity="10"/>
</int:channel>
----
====
====
[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
----
<int:channel id="priorityChannel" datatype="example.Widget">
<int:priority-queue comparator="widgetComparator"
capacity="10"/>
</int:channel>
----
====
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]
----
<int:channel id="rendezvousChannel"/>
<int:rendezvous-queue/>
</int:channel>
----
====
====
[source,java]
[source, java, role="primary"]
.Java
----
@Bean
public PollableChannel rendezvousChannel() {
return new RendezvousChannel();
}
----
[source, xml, role="secondary"]
.XML
----
<int:channel id="rendezvousChannel"/>
<int:rendezvous-queue/>
</int:channel>
----
====
[[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
----
<int:wire-tap pattern="input*, thing2*, thing1" order="3" channel="wiretapChannel"/>
----