SpringCloud/spring-cloud-static
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
Files
gh-pages
spring-cloud-static/Edgware.SR6/multi/multi__programming_model.html
Ryan Baxter 790c2e4e04 Sync docs from vEdgware.SR6 to gh-pages
2019-05-28 19:54:23 -04:00

337 lines
60 KiB
HTML
Raw Permalink Blame History

<html><head>
<meta http-equiv="Content-Type" content="text/html; charset=ISO-8859-1">
<title>26.&nbsp;Programming Model</title><link rel="stylesheet" type="text/css" href="css/manual-multipage.css"><meta name="generator" content="DocBook XSL Stylesheets V1.78.1"><link rel="home" href="multi_spring-cloud.html" title="Spring Cloud"><link rel="up" href="multi__spring_cloud_stream.html" title="Part&nbsp;IV.&nbsp;Spring Cloud Stream"><link rel="prev" href="multi__main_concepts.html" title="25.&nbsp;Main Concepts"><link rel="next" href="multi__binders.html" title="27.&nbsp;Binders"></head><body bgcolor="white" text="black" link="#0000FF" vlink="#840084" alink="#0000FF"><div class="navheader"><table width="100%" summary="Navigation header"><tr><th colspan="3" align="center">26.&nbsp;Programming Model</th></tr><tr><td width="20%" align="left"><a accesskey="p" href="multi__main_concepts.html">Prev</a>&nbsp;</td><th width="60%" align="center">Part&nbsp;IV.&nbsp;Spring Cloud Stream</th><td width="20%" align="right">&nbsp;<a accesskey="n" href="multi__binders.html">Next</a></td></tr></table><hr></div><div class="chapter"><div class="titlepage"><div><div><h2 class="title"><a name="_programming_model" href="#_programming_model"></a>26.&nbsp;Programming Model</h2></div></div></div><p>This section describes Spring Cloud Stream&#8217;s programming model.
Spring Cloud Stream provides a number of predefined annotations for declaring bound input and output channels as well as how to listen to channels.</p><div class="section"><div class="titlepage"><div><div><h2 class="title" style="clear: both"><a name="_declaring_and_binding_channels" href="#_declaring_and_binding_channels"></a>26.1&nbsp;Declaring and Binding Channels</h2></div></div></div><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="_triggering_binding_via_literal_enablebinding_literal" href="#_triggering_binding_via_literal_enablebinding_literal"></a>26.1.1&nbsp;Triggering Binding Via <code class="literal">@EnableBinding</code></h3></div></div></div><p>You can turn a Spring application into a Spring Cloud Stream application by applying the <code class="literal">@EnableBinding</code> annotation to one of the application&#8217;s configuration classes.
The <code class="literal">@EnableBinding</code> annotation itself is meta-annotated with <code class="literal">@Configuration</code> and triggers the configuration of Spring Cloud Stream infrastructure:</p><pre class="programlisting">...
<em><span class="hl-annotation" style="color: gray">@Import(...)</span></em>
<em><span class="hl-annotation" style="color: gray">@Configuration</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableIntegration</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <em><span class="hl-annotation" style="color: gray">@interface</span></em> EnableBinding {
...
Class&lt;?&gt;[] value() <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">default</span> {};
}</pre><p>The <code class="literal">@EnableBinding</code> annotation can take as parameters one or more interface classes that contain methods which represent bindable components (typically message channels).</p><div class="note" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Note"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Note]" src="images/note.png"></td><th align="left">Note</th></tr><tr><td align="left" valign="top"><p>The <code class="literal">@EnableBinding</code> annotation is only required on your <code class="literal">Configuration</code> classes, you can provide as many binding interfaces as you need, for instance: <code class="literal">@EnableBinding(value={Orders.class, Payment.class}</code>.
Where both <code class="literal">Order</code> and <code class="literal">Payment</code> interfaces would declare <code class="literal">@Input</code> and <code class="literal">@Output</code> channels.</p></td></tr></table></div></div><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="__literal_input_literal_and_literal_output_literal" href="#__literal_input_literal_and_literal_output_literal"></a>26.1.2&nbsp;<code class="literal">@Input</code> and <code class="literal">@Output</code></h3></div></div></div><p>A Spring Cloud Stream application can have an arbitrary number of input and output channels defined in an interface as <code class="literal">@Input</code> and <code class="literal">@Output</code> methods:</p><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">interface</span> Barista {
<em><span class="hl-annotation" style="color: gray">@Input</span></em>
SubscribableChannel orders();
<em><span class="hl-annotation" style="color: gray">@Output</span></em>
MessageChannel hotDrinks();
<em><span class="hl-annotation" style="color: gray">@Output</span></em>
MessageChannel coldDrinks();
}</pre><p>Using this interface as a parameter to <code class="literal">@EnableBinding</code> will trigger the creation of three bound channels named <code class="literal">orders</code>, <code class="literal">hotDrinks</code>, and <code class="literal">coldDrinks</code>, respectively.</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Barista.class)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> CafeConfiguration {
...
}</pre><div class="note" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Note"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Note]" src="images/note.png"></td><th align="left">Note</th></tr><tr><td align="left" valign="top"><p>In Spring Cloud Stream, the bindable <code class="literal">MessageChannel</code> components are the Spring Messaging <code class="literal">MessageChannel</code> (for outbound) and its extension <code class="literal">SubscribableChannel</code> (for inbound).
Using the same mechanism other bindable components can be supported.
<code class="literal">KStream</code> support in Spring Cloud Stream Kafka binder is one such example where KStream is used as inbound/outbound <code class="literal">bindable</code> components.
In this documentation, we will continue to refer to MessageChannels as the <code class="literal">bindable</code> components.</p></td></tr></table></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_customizing_channel_names" href="#_customizing_channel_names"></a>Customizing Channel Names</h4></div></div></div><p>Using the <code class="literal">@Input</code> and <code class="literal">@Output</code> annotations, you can specify a customized channel name for the channel, as shown in the following example:</p><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">interface</span> Barista {
...
<em><span class="hl-annotation" style="color: gray">@Input("inboundOrders")</span></em>
SubscribableChannel orders();
}</pre><p>In this example, the created bound channel will be named <code class="literal">inboundOrders</code>.</p></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="__literal_source_literal_literal_sink_literal_and_literal_processor_literal" href="#__literal_source_literal_literal_sink_literal_and_literal_processor_literal"></a><code class="literal">Source</code>, <code class="literal">Sink</code>, and <code class="literal">Processor</code></h4></div></div></div><p>For easy addressing of the most common use cases, which involve either an input channel, an output channel, or both, Spring Cloud Stream provides three predefined interfaces out of the box.</p><p><code class="literal">Source</code> can be used for an application which has a single outbound channel.</p><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">interface</span> Source {
String OUTPUT = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"output"</span>;
<em><span class="hl-annotation" style="color: gray">@Output(Source.OUTPUT)</span></em>
MessageChannel output();
}</pre><p><code class="literal">Sink</code> can be used for an application which has a single inbound channel.</p><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">interface</span> Sink {
String INPUT = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"input"</span>;
<em><span class="hl-annotation" style="color: gray">@Input(Sink.INPUT)</span></em>
SubscribableChannel input();
}</pre><p><code class="literal">Processor</code> can be used for an application which has both an inbound channel and an outbound channel.</p><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">interface</span> Processor <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">extends</span> Source, Sink {
}</pre><p>Spring Cloud Stream provides no special handling for any of these interfaces; they are only provided out of the box.</p></div></div><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="_accessing_bound_channels" href="#_accessing_bound_channels"></a>26.1.3&nbsp;Accessing Bound Channels</h3></div></div></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_injecting_the_bound_interfaces" href="#_injecting_the_bound_interfaces"></a>Injecting the Bound Interfaces</h4></div></div></div><p>For each bound interface, Spring Cloud Stream will generate a bean that implements the interface.
Invoking a <code class="literal">@Input</code>-annotated or <code class="literal">@Output</code>-annotated method of one of these beans will return the relevant bound channel.</p><p>The bean in the following example sends a message on the output channel when its <code class="literal">hello</code> method is invoked.
It invokes <code class="literal">output()</code> on the injected <code class="literal">Source</code> bean to retrieve the target channel.</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@Component</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> SendingBean {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> Source source;
<em><span class="hl-annotation" style="color: gray">@Autowired</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> SendingBean(Source source) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.source = source;
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> sayHello(String name) {
source.output().send(MessageBuilder.withPayload(name).build());
}
}</pre></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_injecting_channels_directly" href="#_injecting_channels_directly"></a>Injecting Channels Directly</h4></div></div></div><p>Bound channels can be also injected directly:</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@Component</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> SendingBean {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> MessageChannel output;
<em><span class="hl-annotation" style="color: gray">@Autowired</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> SendingBean(MessageChannel output) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.output = output;
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> sayHello(String name) {
output.send(MessageBuilder.withPayload(name).build());
}
}</pre><p>If the name of the channel is customized on the declaring annotation, that name should be used instead of the method name.
Given the following declaration:</p><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">interface</span> CustomSource {
...
<em><span class="hl-annotation" style="color: gray">@Output("customOutput")</span></em>
MessageChannel output();
}</pre><p>The channel will be injected as shown in the following example:</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@Component</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> SendingBean {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> MessageChannel output;
<em><span class="hl-annotation" style="color: gray">@Autowired</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> SendingBean(<em><span class="hl-annotation" style="color: gray">@Qualifier("customOutput")</span></em> MessageChannel output) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.output = output;
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> sayHello(String name) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.output.send(MessageBuilder.withPayload(name).build());
}
}</pre></div></div><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="_producing_and_consuming_messages" href="#_producing_and_consuming_messages"></a>26.1.4&nbsp;Producing and Consuming Messages</h3></div></div></div><p>You can write a Spring Cloud Stream application using either Spring Integration annotations or Spring Cloud Stream&#8217;s <code class="literal">@StreamListener</code> annotation.
The <code class="literal">@StreamListener</code> annotation is modeled after other Spring Messaging annotations (such as <code class="literal">@MessageMapping</code>, <code class="literal">@JmsListener</code>, <code class="literal">@RabbitListener</code>, etc.) but adds content type management and type coercion features.</p><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_native_spring_integration_support" href="#_native_spring_integration_support"></a>Native Spring Integration Support</h4></div></div></div><p>Because Spring Cloud Stream is based on Spring Integration, Stream completely inherits Integration&#8217;s foundation and infrastructure as well as the component itself.
For example, you can attach the output channel of a <code class="literal">Source</code> to a <code class="literal">MessageSource</code>:</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Source.class)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> TimerSource {
<em><span class="hl-annotation" style="color: gray">@Value("${format}")</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> String format;
<em><span class="hl-annotation" style="color: gray">@Bean</span></em>
<em><span class="hl-annotation" style="color: gray">@InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> MessageSource&lt;String&gt; timerMessageSource() {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> () -&gt; <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> GenericMessage&lt;&gt;(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> SimpleDateFormat(format).format(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> Date()));
}
}</pre><p>Or you can use a processor&#8217;s channels in a transformer:</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Processor.class)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> TransformProcessor {
<em><span class="hl-annotation" style="color: gray">@Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> Object transform(String message) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> message.toUpperCase();
}
}</pre><div class="note" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Note"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Note]" src="images/note.png"></td><th align="left">Note</th></tr><tr><td align="left" valign="top"><p>It&#8217;s important to understant that when you consume from the same binding using <code class="literal">@StreamListener</code> a pubsub model is used, where each method annotated with <code class="literal">@StreamListener</code> receives it&#8217;s own copy of the message, each one has its own consumer group.
However, if you share a bindable channel as an input for <code class="literal">@Aggregator</code>, <code class="literal">@Transformer</code> or <code class="literal">@ServiceActivator</code>, those will consume in a competing model, no individual consumer group is created for each subscription.</p></td></tr></table></div></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_spring_integration_error_channel_support" href="#_spring_integration_error_channel_support"></a>Spring Integration Error Channel Support</h4></div></div></div><p>Spring Cloud Stream supports publishing error messages received by the Spring Integration global
error channel. Error messages sent to the <code class="literal">errorChannel</code> can be published to a specific destination
at the broker by configuring a binding for the outbound target named <code class="literal">error</code>. For example, to
publish error messages to a broker destination named "myErrors", provide the following property:
<code class="literal">spring.cloud.stream.bindings.error.destination=myErrors</code>.</p></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="binder-error-channels" href="#binder-error-channels"></a>Message Channel Binders and Error Channels</h4></div></div></div><p>Starting with <span class="emphasis"><em>version 1.3</em></span>, some <code class="literal">MessageChannel</code> - based binders publish errors to a discrete error channel for each destination.
In addition, these error channels are bridged to the global Spring Integration <code class="literal">errorChannel</code> mentioned above.
You can therefore consume errors for specific destinations and/or for all destinations, using a standard Spring Integration flow (<code class="literal">IntegrationFlow</code>, <code class="literal">@ServiceActivator</code>, etc).</p><p>On the consumer side, the listener thread catches any exceptions and forwards an <code class="literal">ErrorMessage</code> to the destination&#8217;s error channel.
The payload of the message is a <code class="literal">MessagingException</code> with the normal <code class="literal">failedMessage</code> and <code class="literal">cause</code> properties.
Usually, the raw data received from the broker is included in a header.
For binders that support (and are configured with) a dead letter destination; a <code class="literal">MessagePublishingErrorHandler</code> is subscribed to the channel, and the raw data is forwarded to the dead letter destination.</p><p>On the producer side; for binders that support some kind of async result after publishing messages (e.g. RabbitMQ, Kafka), you can enable an error channel by setting the <code class="literal">&#8230;&#8203;producer.errorChannelEnabled</code> to <code class="literal">true</code>.
The payload of the <code class="literal">ErrorMessage</code> depends on the binder implementation but will be a <code class="literal">MessagingException</code> with the normal <code class="literal">failedMessage</code> property, as well as additional properties about the failure.
Refer to the binder documentation for complete details.</p></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_using_streamlistener_for_automatic_content_type_handling" href="#_using_streamlistener_for_automatic_content_type_handling"></a>Using @StreamListener for Automatic Content Type Handling</h4></div></div></div><p>Complementary to its Spring Integration support, Spring Cloud Stream provides its own <code class="literal">@StreamListener</code> annotation, modeled after other Spring Messaging annotations (e.g. <code class="literal">@MessageMapping</code>, <code class="literal">@JmsListener</code>, <code class="literal">@RabbitListener</code>, etc.).
The <code class="literal">@StreamListener</code> annotation provides a simpler model for handling inbound messages, especially when dealing with use cases that involve content type management and type coercion.</p><p>Spring Cloud Stream provides an extensible <code class="literal">MessageConverter</code> mechanism for handling data conversion by bound channels and for, in this case, dispatching to methods annotated with <code class="literal">@StreamListener</code>.
The following is an example of an application which processes external <code class="literal">Vote</code> events:</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Sink.class)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> VoteHandler {
<em><span class="hl-annotation" style="color: gray">@Autowired</span></em>
VotingService votingService;
<em><span class="hl-annotation" style="color: gray">@StreamListener(Sink.INPUT)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> handle(Vote vote) {
votingService.record(vote);
}
}</pre><p>The distinction between <code class="literal">@StreamListener</code> and a Spring Integration <code class="literal">@ServiceActivator</code> is seen when considering an inbound <code class="literal">Message</code> that has a <code class="literal">String</code> payload and a <code class="literal">contentType</code> header of <code class="literal">application/json</code>.
In the case of <code class="literal">@StreamListener</code>, the <code class="literal">MessageConverter</code> mechanism will use the <code class="literal">contentType</code> header to parse the <code class="literal">String</code> payload into a <code class="literal">Vote</code> object.</p><p>As with other Spring Messaging methods, method arguments can be annotated with <code class="literal">@Payload</code>, <code class="literal">@Headers</code> and <code class="literal">@Header</code>.</p><div class="note" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Note"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Note]" src="images/note.png"></td><th align="left">Note</th></tr><tr><td align="left" valign="top"><p>For methods which return data, you must use the <code class="literal">@SendTo</code> annotation to specify the output binding destination for data returned by the method:</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Processor.class)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> TransformProcessor {
<em><span class="hl-annotation" style="color: gray">@Autowired</span></em>
VotingService votingService;
<em><span class="hl-annotation" style="color: gray">@StreamListener(Processor.INPUT)</span></em>
<em><span class="hl-annotation" style="color: gray">@SendTo(Processor.OUTPUT)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> VoteResult handle(Vote vote) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> votingService.record(vote);
}
}</pre></td></tr></table></div></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_using_streamlistener_for_dispatching_messages_to_multiple_methods" href="#_using_streamlistener_for_dispatching_messages_to_multiple_methods"></a>Using @StreamListener for dispatching messages to multiple methods</h4></div></div></div><p>Since version 1.2, Spring Cloud Stream supports dispatching messages to multiple <code class="literal">@StreamListener</code> methods registered on an input channel, based on a condition.</p><p>In order to be eligible to support conditional dispatching, a method must satisfy the follow conditions:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem">it must not return a value</li><li class="listitem">it must be an individual message handling method (reactive API methods are not supported)</li></ul></div><p>The condition is specified via a SpEL expression in the <code class="literal">condition</code> attribute of the annotation and is evaluated for each message.
All the handlers that match the condition will be invoked in the same thread and no assumption must be made about the order in which the invocations take place.</p><p>An example of using <code class="literal">@StreamListener</code> with dispatching conditions can be seen below.
In this example, all the messages bearing a header <code class="literal">type</code> with the value <code class="literal">foo</code> will be dispatched to the <code class="literal">receiveFoo</code> method, and all the messages bearing a header <code class="literal">type</code> with the value <code class="literal">bar</code> will be dispatched to the <code class="literal">receiveBar</code> method.</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Sink.class)</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableAutoConfiguration</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> TestPojoWithAnnotatedArguments {
<em><span class="hl-annotation" style="color: gray">@StreamListener(target = Sink.INPUT, condition = "headers['type']=='foo'")</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> receiveFoo(<em><span class="hl-annotation" style="color: gray">@Payload</span></em> FooPojo fooPojo) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-comment">// handle the message</span>
}
<em><span class="hl-annotation" style="color: gray">@StreamListener(target = Sink.INPUT, condition = "headers['type']=='bar'")</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> receiveBar(<em><span class="hl-annotation" style="color: gray">@Payload</span></em> BarPojo barPojo) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-comment">// handle the message</span>
}
}</pre><div class="note" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Note"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Note]" src="images/note.png"></td><th align="left">Note</th></tr><tr><td align="left" valign="top"><p>Dispatching via <code class="literal">@StreamListener</code> conditions is only supported for handlers of individual messages, and not for reactive programming support (described below).</p></td></tr></table></div></div></div><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="_reactive_programming_support" href="#_reactive_programming_support"></a>26.1.5&nbsp;Reactive Programming Support</h3></div></div></div><p>Spring Cloud Stream also supports the use of reactive APIs where incoming and outgoing data is handled as continuous data flows.
Support for reactive APIs is available via the <code class="literal">spring-cloud-stream-reactive</code>, which needs to be added explicitly to your project.</p><p>The programming model with reactive APIs is declarative, where instead of specifying how each individual message should be handled, you can use operators that describe functional transformations from inbound to outbound data flows.</p><p>Spring Cloud Stream supports the following reactive APIs:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem">Reactor</li><li class="listitem">RxJava 1.x</li></ul></div><p>In the future, it is intended to support a more generic model based on Reactive Streams.</p><p>The reactive programming model is also using the <code class="literal">@StreamListener</code> annotation for setting up reactive handlers. The differences are that:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem">the <code class="literal">@StreamListener</code> annotation must not specify an input or output, as they are provided as arguments and return values from the method;</li><li class="listitem">the arguments of the method must be annotated with <code class="literal">@Input</code> and <code class="literal">@Output</code> indicating which input or output will the incoming and respectively outgoing data flows connect to;</li><li class="listitem">the return value of the method, if any, will be annotated with <code class="literal">@Output</code>, indicating the input where data shall be sent.</li></ul></div><div class="note" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Note"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Note]" src="images/note.png"></td><th align="left">Note</th></tr><tr><td align="left" valign="top"><p>Reactive programming support requires Java 1.8.</p></td></tr></table></div><div class="note" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Note"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Note]" src="images/note.png"></td><th align="left">Note</th></tr><tr><td align="left" valign="top"><p>As of Spring Cloud Stream 1.1.1 and later (starting with release train Brooklyn.SR2), reactive programming support requires the use of Reactor 3.0.4.RELEASE and higher.
Earlier Reactor versions (including 3.0.1.RELEASE, 3.0.2.RELEASE and 3.0.3.RELEASE) are not supported.
<code class="literal">spring-cloud-stream-reactive</code> will transitively retrieve the proper version, but it is possible for the project structure to manage the version of the <code class="literal">io.projectreactor:reactor-core</code> to an earlier release, especially when using Maven.
This is the case for projects generated via Spring Initializr with Spring Boot 1.x, which will override the Reactor version to <code class="literal">2.0.8.RELEASE</code>.
In such cases you must ensure that the proper version of the artifact is released.
This can be simply achieved by adding a direct dependency on <code class="literal">io.projectreactor:reactor-core</code> with a version of <code class="literal">3.0.4.RELEASE</code> or later to your project.</p></td></tr></table></div><div class="note" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Note"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Note]" src="images/note.png"></td><th align="left">Note</th></tr><tr><td align="left" valign="top"><p>The use of term <code class="literal">reactive</code> is currently referring to the reactive APIs being used and not to the execution model being reactive (i.e. the bound endpoints are still using a 'push' rather than 'pull' model). While some backpressure support is provided by the use of Reactor, we do intend on the long run to support entirely reactive pipelines by the use of native reactive clients for the connected middleware.</p></td></tr></table></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_reactor_based_handlers" href="#_reactor_based_handlers"></a>Reactor-based handlers</h4></div></div></div><p>A Reactor based handler can have the following argument types:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem">For arguments annotated with <code class="literal">@Input</code>, it supports the Reactor type <code class="literal">Flux</code>.
The parameterization of the inbound Flux follows the same rules as in the case of individual message handling: it can be the entire <code class="literal">Message</code>, a POJO which can be the <code class="literal">Message</code> payload, or a POJO which is the result of a transformation based on the <code class="literal">Message</code> content-type header. Multiple inputs are provided;</li><li class="listitem">For arguments annotated with <code class="literal">Output</code>, it supports the type <code class="literal">FluxSender</code> which connects a <code class="literal">Flux</code> produced by the method with an output. Generally speaking, specifying outputs as arguments is only recommended when the method can have multiple outputs;</li></ul></div><p>A Reactor based handler supports a return type of <code class="literal">Flux</code>, case in which it must be annotated with <code class="literal">@Output</code>. We recommend using the return value of the method when a single output flux is available.</p><p>Here is an example of a simple Reactor-based Processor.</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Processor.class)</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableAutoConfiguration</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> UppercaseTransformer {
<em><span class="hl-annotation" style="color: gray">@StreamListener</span></em>
<em><span class="hl-annotation" style="color: gray">@Output(Processor.OUTPUT)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> Flux&lt;String&gt; receive(<em><span class="hl-annotation" style="color: gray">@Input(Processor.INPUT)</span></em> Flux&lt;String&gt; input) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> input.map(s -&gt; s.toUpperCase());
}
}</pre><p>The same processor using output arguments looks like this:</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Processor.class)</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableAutoConfiguration</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> UppercaseTransformer {
<em><span class="hl-annotation" style="color: gray">@StreamListener</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> receive(<em><span class="hl-annotation" style="color: gray">@Input(Processor.INPUT)</span></em> Flux&lt;String&gt; input,
<em><span class="hl-annotation" style="color: gray">@Output(Processor.OUTPUT)</span></em> FluxSender output) {
output.send(input.map(s -&gt; s.toUpperCase()));
}
}</pre></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_rxjava_1_x_support" href="#_rxjava_1_x_support"></a>RxJava 1.x support</h4></div></div></div><p>RxJava 1.x handlers follow the same rules as Reactor-based one, but will use <code class="literal">Observable</code> and <code class="literal">ObservableSender</code> arguments and return types.</p><p>So the first example above will become:</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Processor.class)</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableAutoConfiguration</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> UppercaseTransformer {
<em><span class="hl-annotation" style="color: gray">@StreamListener</span></em>
<em><span class="hl-annotation" style="color: gray">@Output(Processor.OUTPUT)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> Observable&lt;String&gt; receive(<em><span class="hl-annotation" style="color: gray">@Input(Processor.INPUT)</span></em> Observable&lt;String&gt; input) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> input.map(s -&gt; s.toUpperCase());
}
}</pre><p>The second example above will become:</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Processor.class)</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableAutoConfiguration</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> UppercaseTransformer {
<em><span class="hl-annotation" style="color: gray">@StreamListener</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> receive(<em><span class="hl-annotation" style="color: gray">@Input(Processor.INPUT)</span></em> Observable&lt;String&gt; input,
<em><span class="hl-annotation" style="color: gray">@Output(Processor.OUTPUT)</span></em> ObservableSender output) {
output.send(input.map(s -&gt; s.toUpperCase()));
}
}</pre></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_reactive_sources" href="#_reactive_sources"></a>Reactive Sources</h4></div></div></div><p>Spring Cloud Stream reactive support also provides the ability for creating reactive sources through the StreamEmitter annotation.
Using StreamEmitter annotation, a regular source may be converted to a reactive one.
StreamEmitter is a method level annotation that marks a method to be an emitter to outputs declared via EnableBinding.
It is not allowed to use the Input annotation along with StreamEmitter, as the methods marked with this annotation are not listening from any input, rather generating to an output.
Following the same programming model used in StreamListener, StreamEmitter also allows flexible ways of using the Output annotation depending on whether the method has any arguments, return type etc.</p><p>Here are some examples of using StreamEmitter in various styles.</p><p>The following example will emit the "Hello World" message every millisecond and publish to a Flux.
In this case, the resulting messages in Flux will be sent to the output channel of the Source.</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Source.class)</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableAutoConfiguration</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> HelloWorldEmitter {
<em><span class="hl-annotation" style="color: gray">@StreamEmitter</span></em>
<em><span class="hl-annotation" style="color: gray">@Output(Source.OUTPUT)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> Flux&lt;String&gt; emit() {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> Flux.intervalMillis(<span class="hl-number">1</span>)
.map(l -&gt; <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"Hello World"</span>);
}
}</pre><p>Following is another flavor of the same sample as above.
Instead of returning a Flux, this method uses a FluxSender to programmatically send Flux from a source.</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Source.class)</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableAutoConfiguration</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> HelloWorldEmitter {
<em><span class="hl-annotation" style="color: gray">@StreamEmitter</span></em>
<em><span class="hl-annotation" style="color: gray">@Output(Source.OUTPUT)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> emit(FluxSender output) {
output.send(Flux.intervalMillis(<span class="hl-number">1</span>)
.map(l -&gt; <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"Hello World"</span>));
}
}</pre><p>Following is exactly same as the above snippet in functionality and style.
However, instead of using an explicit Output annotation at the method level, it is used as the method parameter level.</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Source.class)</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableAutoConfiguration</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> HelloWorldEmitter {
<em><span class="hl-annotation" style="color: gray">@StreamEmitter</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> emit(<em><span class="hl-annotation" style="color: gray">@Output(Source.OUTPUT)</span></em> FluxSender output) {
output.send(Flux.intervalMillis(<span class="hl-number">1</span>)
.map(l -&gt; <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"Hello World"</span>));
}
}</pre><p>Here is yet another flavor of writing reacting sources using the Reactive Streams Publisher API and the support for it in the <a class="link" href="https://github.com/spring-projects/spring-integration-java-dsl/wiki/Spring-Integration-Java-DSL-Reference" target="_top">Spring Integration Java DSL</a>.
The Publisher is still using Reactor Flux under the hood, but from an application perspective, that is transparent to the user and only needs Reactive Streams and Java DSL for Spring Integration.</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@EnableBinding(Source.class)</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableAutoConfiguration</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> HelloWorldEmitter {
<em><span class="hl-annotation" style="color: gray">@StreamEmitter</span></em>
<em><span class="hl-annotation" style="color: gray">@Output(Source.OUTPUT)</span></em>
<em><span class="hl-annotation" style="color: gray">@Bean</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> Publisher&lt;Message&lt;String&gt;&gt; emit() {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> IntegrationFlows.from(() -&gt;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> GenericMessage&lt;&gt;(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"Hello World"</span>),
e -&gt; e.poller(p -&gt; p.fixedDelay(<span class="hl-number">1</span>)))
.toReactivePublisher();
}
}</pre></div></div><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="_aggregation" href="#_aggregation"></a>26.1.6&nbsp;Aggregation</h3></div></div></div><p>Spring Cloud Stream provides support for aggregating multiple applications together, connecting their input and output channels directly and avoiding the additional cost of exchanging messages via a broker.
As of version 1.0 of Spring Cloud Stream, aggregation is supported only for the following types of applications:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem"><span class="emphasis"><em>sources</em></span> - applications with a single output channel named <code class="literal">output</code>, typically having a single binding of the type <code class="literal">org.springframework.cloud.stream.messaging.Source</code></li><li class="listitem"><span class="emphasis"><em>sinks</em></span> - applications with a single input channel named <code class="literal">input</code>, typically having a single binding of the type <code class="literal">org.springframework.cloud.stream.messaging.Sink</code></li><li class="listitem"><span class="emphasis"><em>processors</em></span> - applications with a single input channel named <code class="literal">input</code> and a single output channel named <code class="literal">output</code>, typically having a single binding of the type <code class="literal">org.springframework.cloud.stream.messaging.Processor</code>.</li></ul></div><p>They can be aggregated together by creating a sequence of interconnected applications, in which the output channel of an element in the sequence is connected to the input channel of the next element, if it exists.
A sequence can start with either a <span class="emphasis"><em>source</em></span> or a <span class="emphasis"><em>processor</em></span>, it can contain an arbitrary number of <span class="emphasis"><em>processors</em></span> and must end with either a <span class="emphasis"><em>processor</em></span> or a <span class="emphasis"><em>sink</em></span>.</p><p>Depending on the nature of the starting and ending element, the sequence may have one or more bindable channels, as follows:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem">if the sequence starts with a source and ends with a sink, all communication between the applications is direct and no channels will be bound</li><li class="listitem">if the sequence starts with a processor, then its input channel will become the <code class="literal">input</code> channel of the aggregate and will be bound accordingly</li><li class="listitem">if the sequence ends with a processor, then its output channel will become the <code class="literal">output</code> channel of the aggregate and will be bound accordingly</li></ul></div><p>Aggregation is performed using the <code class="literal">AggregateApplicationBuilder</code> utility class, as in the following example.
Let&#8217;s consider a project in which we have source, processor and a sink, which may be defined in the project, or may be contained in one of the project&#8217;s dependencies.</p><div class="note" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Note"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Note]" src="images/note.png"></td><th align="left">Note</th></tr><tr><td align="left" valign="top"><p>Each component (source, sink or processor) in an aggregate application must be provided in a separate package if the configuration classes use <code class="literal">@SpringBootApplication</code>.
This is required to avoid cross-talk between applications, due to the classpath scanning performed by <code class="literal">@SpringBootApplication</code> on the configuration classes inside the same package.
In the example below, it can be seen that the Source, Processor and Sink application classes are grouped in separate packages.
A possible alternative is to provide the source, sink or processor configuration in a separate <code class="literal">@Configuration</code> class, avoid the use of <code class="literal">@SpringBootApplication</code>/<code class="literal">@ComponentScan</code> and use those for aggregation.</p></td></tr></table></div><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">package</span> com.app.mysink;
<em><span class="hl-annotation" style="color: gray">@SpringBootApplication</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableBinding(Sink.class)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> SinkApplication {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> Logger logger = LoggerFactory.getLogger(SinkApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>);
<em><span class="hl-annotation" style="color: gray">@ServiceActivator(inputChannel=Sink.INPUT)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> loggerSink(Object payload) {
logger.info(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"Received: "</span> + payload);
}
}</pre><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">package</span> com.app.myprocessor;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-comment">// Imports omitted</span>
<em><span class="hl-annotation" style="color: gray">@SpringBootApplication</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableBinding(Processor.class)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> ProcessorApplication {
<em><span class="hl-annotation" style="color: gray">@Transformer</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> String loggerSink(String payload) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> payload.toUpperCase();
}
}</pre><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">package</span> com.app.mysource;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-comment">// Imports omitted</span>
<em><span class="hl-annotation" style="color: gray">@SpringBootApplication</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableBinding(Source.class)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> SourceApplication {
<em><span class="hl-annotation" style="color: gray">@InboundChannelAdapter(value = Source.OUTPUT)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> String timerMessageSource() {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> SimpleDateFormat().format(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> Date());
}
}</pre><p>Each configuration can be used for running a separate component, but in this case they can be aggregated together as follows:</p><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">package</span> com.app;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-comment">// Imports omitted</span>
<em><span class="hl-annotation" style="color: gray">@SpringBootApplication</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> SampleAggregateApplication {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> main(String[] args) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> AggregateApplicationBuilder()
.from(SourceApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>).args(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"--fixedDelay=5000"</span>)
.via(ProcessorApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>)
.to(SinkApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>).args(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"--debug=true"</span>).run(args);
}
}</pre><p>The starting component of the sequence is provided as argument to the <code class="literal">from()</code> method.
The ending component of the sequence is provided as argument to the <code class="literal">to()</code> method.
Intermediate processors are provided as argument to the <code class="literal">via()</code> method.
Multiple processors of the same type can be chained together (e.g. for pipelining transformations with different configurations).
For each component, the builder can provide runtime arguments for Spring Boot configuration.</p><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_configuring_aggregate_application" href="#_configuring_aggregate_application"></a>Configuring aggregate application</h4></div></div></div><p>Spring Cloud Stream supports passing properties for the individual applications inside the aggregate application using 'namespace' as prefix.</p><p>The namespace can be set for applications as follows:</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@SpringBootApplication</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> SampleAggregateApplication {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> main(String[] args) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> AggregateApplicationBuilder()
.from(SourceApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>).namespace(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"source"</span>).args(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"--fixedDelay=5000"</span>)
.via(ProcessorApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>).namespace(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"processor1"</span>)
.to(SinkApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>).namespace(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"sink"</span>).args(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"--debug=true"</span>).run(args);
}
}</pre><p>Once the 'namespace' is set for the individual applications, the application properties with the <code class="literal">namespace</code> as prefix can be passed to the aggregate application using any supported property source (commandline, environment properties etc.,)</p><p>For instance, to override the default <code class="literal">fixedDelay</code> and <code class="literal">debug</code> properties of 'source' and 'sink' applications:</p><pre class="screen">java -jar target/MyAggregateApplication-0.0.1-SNAPSHOT.jar --source.fixedDelay=10000 --sink.debug=false</pre></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_configuring_binding_service_properties_for_non_self_contained_aggregate_application" href="#_configuring_binding_service_properties_for_non_self_contained_aggregate_application"></a>Configuring binding service properties for non self contained aggregate application</h4></div></div></div><p>The non self-contained aggregate application is bound to external broker via either or both the inbound/outbound components (typically, message channels) of the aggregate application while the applications inside the aggregate application are directly bound.
For example: a source application&#8217;s output and a processor application&#8217;s input are directly bound while the processor&#8217;s output channel is bound to an external destination at the broker.
When passing the binding service properties for non-self contained aggregate application, it is required to pass the binding service properties to the aggregate application instead of setting them as 'args' to individual child application.
For instance,</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@SpringBootApplication</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> SampleAggregateApplication {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> main(String[] args) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> AggregateApplicationBuilder()
.from(SourceApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>).namespace(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"source"</span>).args(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"--fixedDelay=5000"</span>)
.via(ProcessorApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>).namespace(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"processor1"</span>).args(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"--debug=true"</span>).run(args);
}
}</pre><p>The binding properties like <code class="literal">--spring.cloud.stream.bindings.output.destination=processor-output</code> need to be specified as one of the external configuration properties (cmdline arg etc.,).</p></div></div></div></div><div class="navfooter"><hr><table width="100%" summary="Navigation footer"><tr><td width="40%" align="left"><a accesskey="p" href="multi__main_concepts.html">Prev</a>&nbsp;</td><td width="20%" align="center"><a accesskey="u" href="multi__spring_cloud_stream.html">Up</a></td><td width="40%" align="right">&nbsp;<a accesskey="n" href="multi__binders.html">Next</a></td></tr><tr><td width="40%" align="left" valign="top">25.&nbsp;Main Concepts&nbsp;</td><td width="20%" align="center"><a accesskey="h" href="multi_spring-cloud.html">Home</a></td><td width="40%" align="right" valign="top">&nbsp;27.&nbsp;Binders</td></tr></table></div></body></html>
Reference in New Issue View Git Blame Copy Permalink