From 6891d5a2e23fb0267ca5378e449ec5b4024e0228 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 8 Nov 2017 18:53:58 -0500 Subject: [PATCH] INT-4277: Add Java DSL chapter to the Docs JIRA: https://jira.spring.io/browse/INT-4277 *Mostly migration from the https://github.com/spring-projects/spring-integration-java-dsl/wiki/Spring-Integration-Java-DSL-Reference * Add cross links from some appropriate chapters * Fix log message in the `WebSocketInboundChannelAdapter` * Document `IntegrationFlowContext` * Document gateways via `IntegrationFlow` --- .../WebSocketInboundChannelAdapter.java | 4 +- src/reference/asciidoc/aggregator.adoc | 2 + src/reference/asciidoc/channel.adoc | 3 + src/reference/asciidoc/dsl.adoc | 888 ++++++++++++++++++ src/reference/asciidoc/gateway.adoc | 10 +- src/reference/asciidoc/index.adoc | 3 + src/reference/asciidoc/resequencer.adoc | 2 + src/reference/asciidoc/router.adoc | 2 + src/reference/asciidoc/service-activator.adoc | 2 + src/reference/asciidoc/splitter.adoc | 2 + src/reference/asciidoc/whats-new.adoc | 2 +- 11 files changed, 913 insertions(+), 7 deletions(-) create mode 100644 src/reference/asciidoc/dsl.adoc diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java index 3fc50e9afc..9d476b5a56 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java @@ -69,7 +69,7 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport private static final byte[] EMPTY_PAYLOAD = new byte[0]; - private final List defaultConverters = new ArrayList(3); + private final List defaultConverters = new ArrayList<>(3); private ApplicationEventPublisher eventPublisher; @@ -263,7 +263,7 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport private boolean isActive() { if (!this.active) { - logger.warn("MessageProducer '" + this + " 'isn't started to accept WebSocket events."); + logger.warn("MessageProducer '" + this + "' isn't started to accept WebSocket events."); } return this.active; } diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 39dff57604..e574a90ac7 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -285,6 +285,8 @@ See <> below for more information. [[aggregator-config]] ==== Configuring an Aggregator +See <> for configuring an Aggregator in Java DSL. + [[aggregator-xml]] ===== Configuring an Aggregator with XML diff --git a/src/reference/asciidoc/channel.adoc b/src/reference/asciidoc/channel.adoc index f679c6f004..65b68aed86 100644 --- a/src/reference/asciidoc/channel.adoc +++ b/src/reference/asciidoc/channel.adoc @@ -856,3 +856,6 @@ The 'nullChannel' acts like `/dev/null`, simply logging any Message sent to it a Any time you face channel resolution errors for a reply that you don't care about, you can set the affected component's `output-channel` attribute to 'nullChannel' (the name 'nullChannel' is reserved within the application context). The 'errorChannel' is used internally for sending error messages and may be overridden with a custom configuration. This is discussed in greater detail in <>. + + +See also <> in Java DSL chapter for more information about message channel and interceptors. diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc new file mode 100644 index 0000000000..82ec5ae660 --- /dev/null +++ b/src/reference/asciidoc/dsl.adoc @@ -0,0 +1,888 @@ +[[java-dsl]] +== Java DSL + +The Spring Integration JavaConfig and DSL extension provides a set of convenient Builders and a fluent API to configure Spring Integration message flows from Spring `@Configuration` classes. + +[[java-dsl-example]] +=== Example Configurations + +[source,java] +---- +@Configuration +@EnableIntegration +public class MyConfiguration { + + @Bean + public AtomicInteger integerSource() { + return new AtomicInteger(); + } + + @Bean + public IntegrationFlow myFlow() { + return IntegrationFlows.from(integerSource::getAndIncrement, + c -> c.poller(Pollers.fixedRate(100))) + .channel("inputChannel") + .filter((Integer p) -> p > 0) + .transform(Object::toString) + .channel(MessageChannels.queue()) + .get(); + } +} +---- + +As the result after `ApplicationContext` start up Spring Integration endpoints and Message Channels will be created as is the case after XML parsing. +Such configuration can be used to replace XML configuration or along side with it. + +[[java-dsl-introduction]] +=== Introduction + +The Java DSL for Spring Integration is essentially a facade for Spring Integration. +The DSL provides a simple way to embed Spring Integration Message Flows into your application using the fluent `Builder` pattern together with existing Java and Annotation configurations from Spring Framework and Spring Integration as well. +Another useful tool to simplify configuration is Java 8 Lambdas. + +The https://github.com/spring-projects/spring-integration-samples/tree/master/dsl/cafe-dsl[cafe] is a good example of using the DSL. + +The DSL is presented by the `IntegrationFlows` Factory for the `IntegrationFlowBuilder`. +This produces the `IntegrationFlow` component, which should be registered as a Spring bean (`@Bean`). +The builder pattern is used to express arbitrarily complex structures as a hierarchy of methods that may accept Lambdas as arguments. + +The `IntegrationFlowBuilder` just collects integration components (`MessageChannel` s, `AbstractEndpoint` s etc.) in the `IntegrationFlow` bean for further parsing and registration of concrete beans in the application context by the `IntegrationFlowBeanPostProcessor`. + +The Java DSL uses Spring Integration classes directly and bypasses any XML generation and parsing. +However, the DSL offers more than syntactic sugar on top of XML. +One of its most compelling features is the ability to define inline Lambdas to implement endpoint logic, eliminating the need for external classes to implement custom logic. +In some sense, Spring Integration's support for the Spring Expression Language (SpEL) and inline scripting address this, but Java Lambdas are easier and much more powerful. + +[[java-dsl-basics]] +=== DSL Basics + +The `org.springframework.integration.dsl` package contains the `IntegrationFlowBuilder` API mentioned above and a bunch of `IntegrationComponentSpec` implementations which are builders too and provide the fluent API to configure concrete endpoints. +The `IntegrationFlowBuilder` infrastructure provides common http://www.eaipatterns.com[EIP] for message based applications, such as channels, endpoints, pollers and channel interceptors. + +Endpoints are expressed as verbs in the DSL to improve readability. +The following list includes the common DSL method names and the associated EIP endpoint: + +* transform -> `Transformer` +* filter -> `Filter` +* handle -> `ServiceActivator` +* split -> `Splitter` +* aggregate -> `Aggregator` +* route -> `Router` +* bridge -> `Bridge` + +Conceptually, integration processes are constructed by composing these endpoints into one or more message flows. +Note that EIP does not formally define the term 'message flow', but it is useful to think of it as a unit of work that uses well known messaging patterns. +The DSL provides an `IntegrationFlow` component to define a composition of channels and endpoints between them, but now `IntegrationFlow` plays only the configuration role to populate real beans in the application context and isn't used at runtime: + +[source,java] +---- +@Bean +public IntegrationFlow integerFlow() { + return IntegrationFlows.from("input") + .transform(Integer::parseInt) + .get(); +} +---- + +Here we use the `IntegrationFlows` factory to define an `IntegrationFlow` bean using EIP-methods from `IntegrationFlowBuilder`. + +The `transform` method accepts a Lambda as an endpoint argument to operate on the message payload. +The real argument of this method is `GenericTransformer`, hence any out-of-the-box transformers (`ObjectToJsonTransformer`, `FileToStringTransformer` etc.) can be used here. + +Under the covers, `IntegrationFlowBuilder` recognizes the `MessageHandler` and endpoint for that: `MessageTransformingHandler` and `ConsumerEndpointFactoryBean`, respectively. +Let's look at another example: + +[source,java] +---- +@Bean +public IntegrationFlow myFlow() { + return IntegrationFlows.from("input") + .filter("World"::equals) + .transform("Hello "::concat) + .handle(System.out::println) + .get(); +} +---- + +The above example composes a sequence of `Filter -> Transformer -> Service Activator`. +The flow is 'one way', that is it does not provide a a reply message but simply prints the payload to STDOUT. +The endpoints are automatically wired together using direct channels. + +[[java-dsl-channels]] +=== Message Channels + +In addition to the `IntegrationFlowBuilder` with EIP-methods the Java DSL provides a fluent API to configure `MessageChannel` s. +For this purpose the `MessageChannels` builder factory is provided: + +[source,java] +---- +@Bean +public MessageChannel priorityChannel() { + return MessageChannels.priority(this.mongoDbChannelMessageStore, "priorityGroup") + .interceptor(wireTap()) + .get(); +} +---- + +The same `MessageChannels` builder factory can be used in the `channel()` EIP-method from `IntegrationFlowBuilder` to wire endpoints similar to an`input-channel`/`output-channel` pair in the XML configuration. +By default endpoints are wired via `DirectChannel` s where the bean name is based on the pattern: `[IntegrationFlow.beanName].channel#[channelNameIndex]`. +This rule is applied for unnamed channels produced by inline `MessageChannels` builder factory usage, too. +However all `MessageChannels` methods have a `channelId` -aware variant to create the bean names for `MessageChannel` s. +The `MessageChannel` references can be used as well as `beanName`, as bean-method invocations. +Here is a sample with possible variants of `channel()` EIP-method usage: + +[source,java] +---- +@Bean +public MessageChannel queueChannel() { + return MessageChannels.queue().get(); +} + +@Bean +public MessageChannel publishSubscribe() { + return MessageChannels.publishSubscribe().get(); +} + +@Bean +public IntegrationFlow channelFlow() { + return IntegrationFlows.from("input") + .fixedSubscriberChannel() + .channel("queueChannel") + .channel(publishSubscribe()) + .channel(MessageChannels.executor("executorChannel", this.taskExecutor)) + .channel("output") + .get(); +} +---- + +* `from("input")` means: 'find and use the `MessageChannel` with the "input" id, or create one'; +* `fixedSubscriberChannel()` produces an instance of `FixedSubscriberChannel` and registers it with name `channelFlow.channel#0`; +* `channel("queueChannel")` works the same way but, of course, uses an existing "queueChannel" bean; +* `channel(publishSubscribe())` - the bean-method reference; +* `channel(MessageChannels.executor("executorChannel", this.taskExecutor))` the `IntegrationFlowBuilder` unwraps `IntegrationComponentSpec` to the `ExecutorChannel` and registers it as "executorChannel"; +* `channel("output")` - registers the `DirectChannel` bean with "output" name as long as there are no beans with this name. + +Note: the `IntegrationFlow` definition shown above is valid and all of its channels are applied to endpoints with `BridgeHandler` s. + +IMPORTANT: Be careful to use the same inline channel definition via `MessageChannels` factory from different `IntegrationFlow` s. +Even if the DSL parsers register non-existing objects as beans, it can't determine the same object (`MessageChannel`) from different `IntegrationFlow` containers. +This is wrong: + +[source,java] +---- +@Bean +public IntegrationFlow startFlow() { + return IntegrationFlows.from("input") + .transform(...) + .channel(MessageChannels.queue("queueChannel")) + .get(); +} + +@Bean +public IntegrationFlow endFlow() { + return IntegrationFlows.from(MessageChannels.queue("queueChannel")) + .handle(...) + .get(); +} +---- + +You end up with: + +``` +Caused by: java.lang.IllegalStateException: +Could not register object [queueChannel] under bean name 'queueChannel': + there is already object [queueChannel] bound + at o.s.b.f.s.DefaultSingletonBeanRegistry.registerSingleton(DefaultSingletonBeanRegistry.java:129) +``` + +To make it working there is just need to declare `@Bean` for that channel and use its bean-method from different `IntegrationFlow` s. + +[[java-dsl-pollers]] +=== Pollers + +A similar fluent API is provided to configure `PollerMetadata` for `AbstractPollingEndpoint` implementations. +The `Pollers` builder factory can be used to configure common bean definitions or those created from `IntegrationFlowBuilder` EIP-methods: + +[source,java] +---- +@Bean(name = PollerMetadata.DEFAULT_POLLER) +public PollerMetadata poller() { + return Pollers.fixedRate(500).get(); +} +---- + +See `Pollers` and `PollerSpec` Java Docs for more information. + +[[java-dsl-endpoints]] +=== DSL and Endpoint Configuration + +All `IntegrationFlowBuilder` EIP-methods have a variant to apply the Lambda parameter to provide options for `AbstractEndpoint` s: `SmartLifecycle`, `PollerMetadata`, `request-handler-advice-chain` etc. +Each of them has generic arguments, so it allows you to simply configure an endpoint and even its `MessageHandler` in the context: + +[source,java] +---- +@Bean +public IntegrationFlow flow2() { + return IntegrationFlows.from(this.inputChannel) + .transform(new PayloadSerializingTransformer(), + c -> c.autoStartup(false).id("payloadSerializingTransformer")) + .transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice())) + .get(); +} +---- + +In addition the `EndpointSpec` provides an `id()` method to allow you to register an endpoint bean with a given bean name, rather than a generated one. + +[[java-dsl-transformers]] +=== Transformers + +The DSL API provides a convenient, fluent `Transformers` factory to be used as inline target object definition within `.transform()` EIP-method: + +[source,java] +---- +@Bean +public IntegrationFlow transformFlow() { + return IntegrationFlows.from("input") + .transform(Transformers.fromJson(MyPojo.class)) + .transform(Transformers.serializer()) + .get(); +} +---- + +It avoids inconvenient coding using setters and makes the flow definition more straightforward. +Note, that `Transformers` can be use to declare target `Transformer` s as `@Bean` s and, again, use them from `IntegrationFlow` definition as bean-methods. +Nevertheless, the DSL parser takes care about bean declarations for inline objects, if they aren't defined as beans yet. + +See `Transformers` Java Docs for more information and supported factory methods. + +[[java-dsl-inbound-adapters]] +=== Inbound Channel Adapters + +Typically message flows start from some Inbound Channel Adapter (e.g. ``). +The adapter is configured with `` and it asks a `MessageSource` for producing messages periodically. +Java DSL allows to start `IntegrationFlow` from a `MessageSource`, too. +For this purpose `IntegrationFlows` builder factory provides overloaded `IntegrationFlows.from(MessageSource messageSource)` method. +The `MessageSource` may be configured as a bean and provided as argument for that method. +The second parameter of `IntegrationFlows.from()` is a `Consumer` Lambda and allows to provide options for the `SourcePollingChannelAdapter`, e.g. `PollerMetadata` or `SmartLifecycle`: + +[source,java] +---- +@Bean +public MessageSource jdbcMessageSource() { + return new JdbcPollingChannelAdapter(this.dataSource, "SELECT * FROM foo"); +} + +@Bean +public IntegrationFlow pollingFlow() { + return IntegrationFlows.from(jdbcMessageSource(), + c -> c.poller(Pollers.fixedRate(100).maxMessagesPerPoll(1))) + .transform(Transformers.toJson()) + .channel("furtherProcessChannel") + .get(); +} +---- + +There is also an `IntegrationFlows.from()` variant based on the `java.util.function.Supplier` if there is no requirements to build `Message` objects directly. +The result of the `Supplier.get()` is wrapped to the `Message` (if it isn't message already) by Framework automatically. + +The next sections discuss selected endpoints which require further explanation. + +[[java-dsl-routers]] +=== Message Routers + +Spring Integration natively provides specialized router types including: + +* `HeaderValueRouter` +* `PayloadTypeRouter` +* `ExceptionTypeRouter` +* `RecipientListRouter` +* `XPathRouter` + +As with many other DSL `IntegrationFlowBuilder` EIP-methods the `route()` method can apply any out-of-the-box `AbstractMessageRouter` implementation, or for convenience a `String` as a SpEL expression, or a `ref`/`method` pair. +In addition `route()` can be configured with a Lambda - the inline method invocation case, and with a Lambda for a `Consumer>`. +The fluent API also provides `AbstractMappingMessageRouter` options like `channelMapping(String key, String channelName)` pairs: + +[source,java] +---- +@Bean +public IntegrationFlow routeFlow() { + return IntegrationFlows.from("routerInput") + .route(p -> p % 2 == 0, + m -> m.suffix("Channel") + .channelMapping("true", "even") + .channelMapping("false", "odd") + ) + .get(); +} +---- + +A simple expression-based router: + +[source,java] +---- +@Bean +public IntegrationFlow routeFlow() { + return IntegrationFlows.from("routerInput") + .route("headers['destChannel']") + .get(); +} +---- + +The `routeToRecipients()` method takes a `Consumer`: + +[source,java] +---- +@Bean +public IntegrationFlow recipientListFlow() { + return IntegrationFlows.from("recipientListInput") + .transform(p -> p.replaceFirst("Payload", "")) + .routeToRecipients(r -> r + .recipient("foo-channel", "'foo' == payload") + .recipient("bar-channel", m -> + m.getHeaders().containsKey("recipient") + && (boolean) m.getHeaders().get("recipient")) + .recipientFlow("'foo' == payload or 'bar' == payload or 'baz' == payload", + f -> f.transform(String::toUpperCase) + .channel(c -> c.queue("recipientListSubFlow1Result"))) + .recipientFlow((String p) -> p.startsWith("baz"), + f -> f.transform("Hello "::concat) + .channel(c -> c.queue("recipientListSubFlow2Result"))) + .recipientFlow(new FunctionExpression>(m -> + "bax".equals(m.getPayload())), + f -> f.channel(c -> c.queue("recipientListSubFlow3Result"))) + .defaultOutputToParentFlow()) + .get(); +} +---- + +The `.defaultOutputToParentFlow()` of the `.routeToRecipients()` allows to make the router's `defaultOutput` as a gateway to continue a process for the unmatched messages in the main flow. + +[[java-dsl-splitters]] +=== Splitters + +A splitter is created using the `split()` EIP-method. +By default, if the payload is a `Iterable`, `Iterator`, `Array`, `Stream` or Reactive `Publisher`, this will output each item as an individual message. +This takes a Lambda, SpEL expression, any `AbstractMessageSplitter` implementation, or can be used without parameters to provide the `DefaultMessageSplitter`. +For example: + +[source,java] +---- +@Bean +public IntegrationFlow splitFlow() { + return IntegrationFlows.from("splitInput") + .split(s -> + s.applySequence(false).get().getT2().setDelimiters(",")) + .channel(MessageChannels.executor(this.taskExecutor())) + .get(); +} +---- + +This creates a splitter that splits a message containing a comma delimited String. +Note: the `getT2()` method comes from `Tuple` `Collection` which is the result of `EndpointSpec.get()` and represents a pair of `ConsumerEndpointFactoryBean` and `DefaultMessageSplitter` for the example above. + +[[java-dsl-aggregators]] +=== Aggregators and Resequencers + +An `Aggregator` is conceptually the converse of a `Splitter`. +It aggregates a sequence of individual messages into a single message and is necessarily more complex. +By default, an aggregator will return a message containing a collection of payloads from incoming messages. +The same rules are applied for the `Resequencer`: + +[source,java] +---- +@Bean +public IntegrationFlow splitAggregateFlow() { + return IntegrationFlows.from("splitAggregateInput") + .split() + .channel(MessageChannels.executor(this.taskExecutor())) + .resequence() + .aggregate() + .get(); +} +---- + +The above is a canonical example of splitter/aggregator pattern. +The `split()` method splits the list into individual messages and sends them to the `ExecutorChannel`. +The `resequence()` method reorders messages by sequence details from message headers. +The `aggregate()` method just collects those messages to the result list. + +However, you may change the default behavior by specifying a release strategy and correlation strategy, among other things. +Consider the following: + +[source,java] +---- +.aggregate(a -> + a.correlationStrategy(m -> m.getHeaders().get("myCorrelationKey")) + .releaseStrategy(g -> g.size() > 10) + .messageStore(messageStore())) +---- + +The similar Lambda configurations are provided for the `resequence()` EIP-method. + +[[java-dsl-handle]] +=== ServiceActivators (.handle()) + +The `.handle()` EIP-method's goal is to invoke any `MessageHandler` implementation or any method on some POJO. +Another option to define "activity" via Lambda expression. +Hence a generic `GenericHandler

` functional interface has been introduced. +Its `handle` method requires two arguments - `P payload` and `Map headers`. +Having that we can define a flow like this: + +[source,java] +---- +@Bean +public IntegrationFlow myFlow() { + return IntegrationFlows.from("flow3Input") + .handle((p, h) -> p * 2) + .get(); +} +---- + +However one main goal of Spring Integration an achieving of `loose coupling` via runtime type conversion from message payload to target arguments of message handler. +Since Java doesn't support generic type resolution for Lambda classes, we introduced a workaround with additional `payloadType` argument for the most EIP-methods and `LambdaMessageProcessor`, which delegates the hard conversion work to the Spring's `ConversionService` using provided `type` and requested message to target method arguments. +The `IntegrationFlow` might look like this: + +[source,java] +---- +@Bean +public IntegrationFlow integerFlow() { + return IntegrationFlows.from("input") + .transform(p - > new String(p, "UTF-8")) + .handle(Integer.class, (p, h) -> p * 2) + .get(); +} +---- + +Of course we register some custom `BytesToIntegerConverter` within `ConversionService` and get rid of that additional `.transform()`. + +[[java-dsl-log]] +=== Operator log() + +For convenience to log the message journey throw the Spring Integration flow (``), a `log()` operator is presented. +Underneath it is represented by the `WireTap` `ChannelInterceptor` and `LoggingHandler` as subscriber. +It is responsible to log message incoming into the next endpoint: + +[source,java] +---- +.filter(...) +.log(LoggingHandler.Level.ERROR, "test.category", m -> m.getHeaders().getId()) +.route(...) +---- + +In this example an `id` header will be logged with `ERROR` level onto "test.category" only for messages passed the filter and before routing. + +[[java-dsl-wiretap]] +=== MessageChannelSpec.wireTap() + +A `.wireTap()` fluent API exists for `MessageChannelSpec` builders. +A target configuration gains much more from Java DSL usage: + +[source,java] +---- +@Bean +public QueueChannelSpec myChannel() { + return MessageChannels.queue() + .wireTap("loggingFlow.input"); +} + +@Bean +public IntegrationFlow loggingFlow() { + return f -> f.log(); +} +---- + +[[java-dsl-flows]] +=== Working With Message Flows + +As we have seen, `IntegrationFlowBuilder` provides a top level API to produce Integration components wired to message flows. +This is convenient if your integration may be accomplished with a single flow (which is often the case). +Alternately `IntegrationFlow` s can be joined via `MessageChannel` s. + +By default, the *MessageFlow* behaves as a *Chain* in Spring Integration parlance. +That is, the endpoints are automatically wired implicitly via `DirectChannel` s. +The message flow is not actually constructed as a chain, affording much more flexibility. +For example, you may send a message to any component within the flow, if you know its `inputChannel` name, i.e., explicitly define it. +You may also reference externally defined channels within a flow to allow the use of channel adapters to enable remote transport protocols, file I/O, and the like, instead of direct channels. +As such, the DSL does not support the Spring Integration *chain* element since it doesn't add much value. + +Since the Spring Integration Java DSL produces the same bean definition model as any other configuration options and is based on the existing Spring Framework `@Configuration` infrastructure, it can be used together with Integration XML definitions and wired with Spring Integration Messaging Annotations configuration. + +Another alternative to define *direct* `IntegrationFlow` s is based on a fact that `IntegrationFlow` can be declared as **Lambda** too: + +[source,java] +---- +@Bean +public IntegrationFlow lambdaFlow() { + return f -> f.filter("World"::equals) + .transform("Hello "::concat) + .handle(System.out::println); +} +---- + +The result of this definition is the same bunch of Integration components wired with implicit direct channel. +Only limitation is here, that this flow is started with named direct channel - `lambdaFlow.input`. +And Lambda flow can't start from `MessageSource` or `MessageProducer`. + +[[java-dsl-function-expression]] +=== FunctionExpression + +The `FunctionExpression` (an implementation of SpEL `Expression`) has been introduced to get a gain of Java and Lambda usage for the method and its `generics` context. +The `Function` option is provided for the DSL components alongside with `expression` option, when there is the implicit `Strategy` variant from Core Spring Integration. +The usage may look like: + +[source,java] +---- +.enrich(e -> e.requestChannel("enrichChannel") + .requestPayload(Message::getPayload) + .propertyFunction("date", m -> new Date())) +---- + +The `FunctionExpression` also supports runtime type conversion as it is done in the standard `SpelExpression`. + +[[java-dsl-subflows]] +=== Sub Flows support + +Some of `if...else` and `publish-subscribe` components provide the support to specify their logic or mapping using **Sub Flows**. +The simplest sample is `.publishSubscribeChannel()`: + +[source,java] +---- +@Bean +public IntegrationFlow subscribersFlow() { + return flow -> flow + .publishSubscribeChannel(Executors.newCachedThreadPool(), s -> s + .subscribe(f -> f + .handle((p, h) -> p / 2) + .channel(c -> c.queue("subscriber1Results"))) + .subscribe(f -> f + .handle((p, h) -> p * 2) + .channel(c -> c.queue("subscriber2Results")))) + .handle((p, h) -> p * 3) + .channel(c -> c.queue("subscriber3Results")); +} +---- + +Of course the same result we can achieve with separate `IntegrationFlow` `@Bean` definitions, but we hope you'll find the subflow style of logic composition useful. + +Similar `publish-subscribe` subflow composition provides `.routeToRecipients()`. + +Another sample is `.discardFlow()` on the `.filter()` instead of `.discardChannel()`. + +The `.route()` deserves special attention. +As a sample: + +[source,java] +---- +@Bean +public IntegrationFlow routeFlow() { + return f -> f + .route(p -> p % 2 == 0, + m -> m.channelMapping("true", "evenChannel") + .subFlowMapping("false", sf -> + sf.handle((p, h) -> p * 3))) + .transform(Object::toString) + .channel(c -> c.queue("oddChannel")); +} +---- + +The `.channelMapping()` continues to work as in regular `Router` mapping, but the `.subFlowMapping()` tied that subflow with main flow. +In other words, any router's subflow returns to the main flow after `.route()`. + +Of course, subflows can be nested with any depth, but we don't recommend to do that because, in fact, even in the router case, adding complex subflows within a flow would quickly begin to look like a plate of spaghetti and difficult for a human to parse. + +[[java-dsl-protocol-adapters]] +=== Using Protocol Adapters + +All of the examples so far illustrate how the DSL supports a messaging architecture using the Spring Integration programming model, but we haven't done any real integration yet. +This requires access to remote resources via http, jms, amqp, tcp, jdbc, ftp, smtp, and the like, or access to the local file system. +Spring Integration supports all of these and more. +Ideally, the DSL should offer first class support for all of them but it is a daunting task to implement all of these and keep up as new adapters are added to Spring Integration. +So the expectation is that the DSL will continually be catching up with Spring Integration. + +Anyway we are providing the hi-level API to define protocol-specific seamlessly. +This is achieved with **Factory** and **Builder** patterns and, of course, with Lambdas. +The factory classes can be considered "Namespace Factories", because they play the same role as XML namespace for components from the concrete protocol-specific Spring Integration modules. +Currently, Spring Integration Java DSL supports `Amqp`, `Feed`, `Jms`, `Files`, `(S)Ftp`, `Http`, `JPA`, `MongoDb`, `TCP/UDP`, `Mail`, `WebFlux` and `Scripts` namespace factories: + +[source,java] +---- +@Bean +public IntegrationFlow amqpFlow() { + return IntegrationFlows.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue())) + .transform("hello "::concat) + .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(); +} +---- + +We show here the usage of namespace factories as inline adapters declarations, however they can be used 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'd also appreciate some prioritization for which adapters/gateways we should support next. + +See more Java DSL samples in the protocol-specific chapter throughout this reference manual. + +All other protocol channel adapters may be configured as generic beans and wired to the `IntegrationFlow`: + +[source,java] +---- +@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; +} +---- + +[[java-dsl-flow-adapter]] +=== IntegrationFlowAdapter + +The `IntegrationFlow` as an interface can be implemented directly and specified as component for scanning: + +[source,java] +---- +@Component +public class MyFlow implements IntegrationFlow { + + @Override + public void configure(IntegrationFlowDefinition f) { + f.transform(String::toUpperCase); + } + +} +---- + +And yes, it is picked up by the `IntegrationFlowBeanPostProcessor` and correctly parsed and registered in the application context. + +For convenience and loosely coupled architecture the `IntegrationFlowAdapter` base class implementation is provided. +It requires a `buildFlow()` method implementation to produce an `IntegrationFlowDefinition` using one of `from()` support methods: + +[source,java] +---- +@Component +public class MyFlowAdapter extends IntegrationFlowAdapter { + + private final AtomicBoolean invoked = new tomicBoolean(); + + 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("foo", "FOO")) + .filter(this) + .handle(this) + .channel(c -> c.queue("myFlowAdapterOutput")); + } + + public String messageSource() { + return "B,A,R"; + } + + @Splitter + public String[] split(String payload) { + return StringUtils.commaDelimitedListToStringArray(payload); + } + + @Transformer + public String transform(String payload) { + return payload.toLowerCase(); + } + + @Aggregator + public String aggregate(List payloads) { + return payloads.stream().collect(Collectors.joining()); + } + + @Filter + public boolean filter(@Header Optional foo) { + return foo.isPresent(); + } + + @ServiceActivator + public String handle(String payload, @Header String foo) { + return payload + ":" + foo; + } + +} +---- + +[[java-dsl-runtime-flows]] +=== Dynamic and runtime Integration Flows + +The `IntegrationFlow` s and therefore all its dependant components can be registered at runtime. +This was done previously by the `BeanFactory.registerSingleton()` hook and now via newly introduced in the Spring Framework `5.0` programmatic `BeanDefinition` registration with the `instanceSupplier` hook: + +[source,java] +---- +BeanDefinition beanDefinition = + BeanDefinitionBuilder.genericBeanDefinition((Class) bean.getClass(), () -> bean) + .getRawBeanDefinition(); + +((BeanDefinitionRegistry) this.beanFactory).registerBeanDefinition(beanName, beanDefinition); +---- + +and 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: + +[source,java] +---- +@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 multi configuration options and have to create several instances of similar flows. +So, we can iterate our options and create and register `IntegrationFlow` s within loop. +Another variant when our source of data isn't Spring-based and we must create it on the fly. +Such a sample is Reactive Streams event source: + +[source,java] +---- +Flux> messageFlux = + Flux.just("1,2,3,4") + .map(v -> v.split(",")) + .flatMapIterable(Arrays::asList) + .map(Integer::parseInt) + .map(GenericMessage::new); + +QueueChannel resultChannel = new QueueChannel(); + +IntegrationFlow integrationFlow = + IntegrationFlows.from(messageFlux) + .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 also for additional, non Integration beans registration. +Usually those additional beans are connection factories (AMQP, JMS, (S)FTP, TCP/UDP etc.), serializers/deserializers or any other required support components. + +Such a dynamically registered `IntegrationFlow` and all its dependant beans can be removed afterwards using `IntegrationFlowRegistration.destroy()` callback. +See `IntegrationFlowContext` JavaDocs for more information. + +[[java-dsl-gateway]] +=== IntegrationFlow as Gateway + +The `IntegrationFlow` can start from the service interface providing `GatewayProxyFactoryBean` component: + +[source,java] +---- +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`. +The service interface can be marked with the `@MessagingGateway` as well as methods with the `@Gateway` annotations. +Nevertheless the `requestChannel` is ignored and overridden with that internal channel for the next component in the `IntegrationFlow`. +Otherwise such a configuration via `IntegrationFlow` won't make sense. + +By default a `GatewayProxyFactoryBean` gets a conventional bean name like `[FLOW_BEAN_NAME.gateway]`. +That id can be changed via `@MessagingGateway.name()` attribute or the overloaded `from(Class serviceInterface, String beanName)` factory method. + +With the Java 8 on board we even can create such an Integration Gateway with the `java.util.function` interfaces: + +[source,java] +---- +@Bean +public IntegrationFlow errorRecovererFlow() { + return IntegrationFlows.from(Function.class, "errorRecovererFunction") + .handle((GenericHandler) (p, h) -> { + throw new RuntimeException("intentional"); + }, e -> e.advice(retryAdvice())) + .get(); +} +---- + +That can be used lately as: + +[source,java] +---- +@Autowired +@Qualifier("errorRecovererFunction") +private Function errorRecovererFlowGateway; +---- diff --git a/src/reference/asciidoc/gateway.adoc b/src/reference/asciidoc/gateway.adoc index 11de768e33..20f43814d1 100644 --- a/src/reference/asciidoc/gateway.adoc +++ b/src/reference/asciidoc/gateway.adoc @@ -440,7 +440,7 @@ The reply timeout is how long the gateway will wait for a reply, or return `null The timeouts can be set as defaults for all methods on the gateway (`defaultRequestTimeout`, `defaultReplyTimeout`) (or on the `MessagingGateway` interface annotation). Individual methods can override these defaults (in `` child elements) or on the `@Gateway` annotation. -Starting with _version 5.0_ the timeouts can be defined as expressions: +Starting with _version 5.0_, the timeouts can be defined as expressions: [source, java] ---- @@ -574,7 +574,7 @@ public interface NoExecGateway { } ---- -IMPORTANT: If the return type is a specific concrete `Future` implementation or some other subinterface that is not supported by the configured executor, the flow will run on the caller's thread and the flow must return the required type in the reply message payload. +IMPORTANT: If the return type is a specific concrete `Future` implementation or some other sub-interface that is not supported by the configured executor, the flow will run on the caller's thread and the flow must return the required type in the reply message payload. [[gw-completable-future]] ===== CompletableFuture @@ -669,7 +669,7 @@ A `Mono` can be used to retrieve the result later (similar to a `Future`) or IMPORTANT: The `Mono` isn't _flushed_ immediately by the framework. Hence the underlying message flow won't be started before the gateway method returns (as it is with `Future` `Executor` task). The flow will be started when the `Mono` is _subscribed_. -Alternatively, the `Mono` (being a `Composable`) might be a part of Reactor stream, when the `subscribe()` is related to the entire `Flux`. +Alternatively, the `Mono` (being a *Composable*) might be a part of Reactor stream, when the `subscribe()` is related to the entire `Flux`. For example: [source,java] @@ -759,10 +759,12 @@ In other words, it will help to ensure a timely response from the Gateway method NOTE: * _reply-timeout_ is unbounded for __ elements (created by the GatewayProxyFactoryBean). Inbound gateways for external integration (ws, http, etc.) share many characteristics and attributes with these gateways. However, for those inbound gateways, the default _reply-timeout_ is 1000 milliseconds (1 second). -If a downstream async handoff is made to another thread, you may need to increase this attribute to allow enough time for the flow to complete before the gateway times out. +If a downstream async hand-off is made to another thread, you may need to increase this attribute to allow enough time for the flow to complete before the gateway times out. IMPORTANT: It is important to understand that the timer starts when the thread returns to the gateway, i.e. when the flow completes or a message is handed off to another thread. At that time, the calling thread starts waiting for the reply. If the flow was completely synchronous, the reply will be immediately available; for asynchronous flows, the thread will wait for up to this time. + +Also see <> in the Java DSL chapter for options to define gateways via `IntegrationFlows`. diff --git a/src/reference/asciidoc/index.adoc b/src/reference/asciidoc/index.adoc index 13c402e74d..11f577a2ff 100644 --- a/src/reference/asciidoc/index.adoc +++ b/src/reference/asciidoc/index.adoc @@ -44,7 +44,10 @@ include::./message-transformation.adoc[] include::./messaging-endpoints.adoc[] +include::./dsl.adoc[] + include::./system-management.adoc[] + [[spring-integration-endpoints]] = Integration Endpoints diff --git a/src/reference/asciidoc/resequencer.adoc b/src/reference/asciidoc/resequencer.adoc index 2eeab2b9e3..d04f93baf9 100644 --- a/src/reference/asciidoc/resequencer.adoc +++ b/src/reference/asciidoc/resequencer.adoc @@ -18,6 +18,8 @@ If you have a large number of disjoint sequences with many gaps, you may experie ==== Configuring a Resequencer +See <> for configuring a Resequencer in Java DSL. + Configuring a resequencer requires only including the appropriate element in XML. A sample resequencer configuration is shown below. diff --git a/src/reference/asciidoc/router.adoc b/src/reference/asciidoc/router.adoc index 7b75106f30..ab4de54972 100644 --- a/src/reference/asciidoc/router.adoc +++ b/src/reference/asciidoc/router.adoc @@ -1110,6 +1110,8 @@ public List route(@Header("orderStatus") OrderStatus status) NOTE: For routing of XML-based Messages, including XPath support, see <>. +Also see <> in Java DSL chapter for more information about routers configuration. + [[dynamic-routers]] ==== Dynamic Routers diff --git a/src/reference/asciidoc/service-activator.adoc b/src/reference/asciidoc/service-activator.adoc index fedd8580ab..06662a9abd 100644 --- a/src/reference/asciidoc/service-activator.adoc +++ b/src/reference/asciidoc/service-activator.adoc @@ -132,6 +132,8 @@ For simple scenarios your _Service Activators_ do not even have to reference a b In the above configuration our service logic is to simply multiply the payload value by 2, and SpEL lets us handle it relatively easy. +See <> in Java DSL chapter for more information about configuring Service Activator. + [[async-service-activator]] ==== Asynchronous Service Activator diff --git a/src/reference/asciidoc/splitter.adoc b/src/reference/asciidoc/splitter.adoc index 02782188a7..a09c0aeddd 100644 --- a/src/reference/asciidoc/splitter.adoc +++ b/src/reference/asciidoc/splitter.adoc @@ -141,3 +141,5 @@ List extractItems(Order order) { ---- Also see <>. + +Also see <> in Java DSL chapter. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index f353c7f887..447cb61c1b 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -13,7 +13,7 @@ development process. The separate https://github.com/spring-projects/spring-integration-java-dsl[Spring Integration Java DSL] project has now been merged into the core Spring Integration project. The `IntegrationComponentSpec` implementations for channel adapters and gateways are distributed to their specific modules. -See <> for more information about Java DSL support for target protocol integration. +See <> for more information about Java DSL support. Also see the https://github.com/spring-projects/spring-integration/wiki/Spring-Integration-4.3-to-5.0-Migration-Guide#java-dsl[4.3 to 5.0 Migration Guide] for the required steps to move to Spring Integration 5.0. ==== Testing Support