From 6f411c8e822d90c50c6cbf5a2a7a61a19cf7f9b1 Mon Sep 17 00:00:00 2001 From: buildmaster Date: Thu, 24 Oct 2019 12:21:48 +0000 Subject: [PATCH] Sync docs from master to gh-pages --- reference/html/README.html | 26 +- reference/html/binders.html | 2 +- reference/html/index.html | 2 +- reference/html/preface.html | 26 +- reference/html/sagan-index.html | 2 +- reference/html/spring-cloud-stream.html | 778 +++++++++++++++--------- 6 files changed, 543 insertions(+), 293 deletions(-) diff --git a/reference/html/README.html b/reference/html/README.html index 71d1abde2..baf049e8b 100644 --- a/reference/html/README.html +++ b/reference/html/README.html @@ -4,7 +4,7 @@ - + Preface @@ -229,7 +229,7 @@ If you chose RabbitMQ for the middleware, your Spring Initializr should now be a -
+
spring initializr
@@ -384,7 +384,10 @@ For something more predictable, you can use an explicit group name by setting

New Features and Components

-

TBD

+

TBD +- Routing Function [Details to follow] +- Multiple bindings with functions [Details to follow] +- Functions with multiple inputs/outputs

@@ -395,8 +398,21 @@ For something more predictable, you can use an explicit group name by setting

Notable Deprecations

-
-

TBD

+
+
    +
  • +

    Reactive module in favor of native support via spring-cloud-function. [Details to follow]

    +
    • +
  • +

    Test support module with MessageCollector [Details to follow]

    +
    • +
  • +

    @StreamMessageConverter [Details to follow]

    +
    • +
  • +

    Original content type - removed

    +
    • +
diff --git a/reference/html/binders.html b/reference/html/binders.html index 6966f8851..93190de9d 100644 --- a/reference/html/binders.html +++ b/reference/html/binders.html @@ -4,7 +4,7 @@ - + Binder Implementations diff --git a/reference/html/index.html b/reference/html/index.html index 736db644a..3edd6666d 100644 --- a/reference/html/index.html +++ b/reference/html/index.html @@ -4,7 +4,7 @@ - + Spring Cloud Stream Reference Documentation diff --git a/reference/html/preface.html b/reference/html/preface.html index 6b0907c4d..73a792451 100644 --- a/reference/html/preface.html +++ b/reference/html/preface.html @@ -4,7 +4,7 @@ - + A Brief History of Spring’s Data Integration Journey @@ -188,7 +188,7 @@ If you chose RabbitMQ for the middleware, your Spring Initializr should now be a
-
+
spring initializr
@@ -343,7 +343,10 @@ For something more predictable, you can use an explicit group name by setting

New Features and Components

-

TBD

+

TBD +- Routing Function [Details to follow] +- Multiple bindings with functions [Details to follow] +- Functions with multiple inputs/outputs

@@ -354,8 +357,21 @@ For something more predictable, you can use an explicit group name by setting

Notable Deprecations

-
-

TBD

+
+
    +
  • +

    Reactive module in favor of native support via spring-cloud-function. [Details to follow]

    +
    • +
  • +

    Test support module with MessageCollector [Details to follow]

    +
    • +
  • +

    @StreamMessageConverter [Details to follow]

    +
    • +
  • +

    Original content type - removed

    +
    • +
diff --git a/reference/html/sagan-index.html b/reference/html/sagan-index.html index a2c3372b9..40c86e8af 100644 --- a/reference/html/sagan-index.html +++ b/reference/html/sagan-index.html @@ -4,7 +4,7 @@ - + Binder Implementations diff --git a/reference/html/spring-cloud-stream.html b/reference/html/spring-cloud-stream.html index a45c6909f..6dab34611 100644 --- a/reference/html/spring-cloud-stream.html +++ b/reference/html/spring-cloud-stream.html @@ -4,7 +4,7 @@ - + Preface @@ -136,21 +136,36 @@ $(addBlockSwitches);
  • Programming Model
    • -
    +
    spring initializr
    @@ -497,7 +513,10 @@ For something more predictable, you can use an explicit group name by setting

    New Features and Components

    -

    TBD

    +

    TBD +- Routing Function [Details to follow] +- Multiple bindings with functions [Details to follow] +- Functions with multiple inputs/outputs

    @@ -508,8 +527,21 @@ For something more predictable, you can use an explicit group name by setting

    Notable Deprecations

    -
    -

    TBD

    +
    +
      +
    • +

      Reactive module in favor of native support via spring-cloud-function. [Details to follow]

      +
      • +
    • +

      Test support module with MessageCollector [Details to follow]

      +
      • +
    • +

      @StreamMessageConverter [Details to follow]

      +
      • +
    • +

      Original content type - removed

      +
      • +
    @@ -605,7 +637,7 @@ This section gives an overview of the following:

    The application communicates with the outside world through input and output channels injected into it by Spring Cloud Stream. Channels are connected to external brokers through middleware-specific Binder implementations.

    -
    +
    SCSt with binder
    @@ -623,8 +655,7 @@ To run a Spring Cloud Stream application in production, you can create an execut

    The Binder Abstraction

    Spring Cloud Stream provides Binder implementations for Kafka and Rabbit MQ. -Spring Cloud Stream also includes a TestSupportBinder, which leaves a channel unmodified so that tests can interact with channels directly and reliably assert on what is received. -You can also use the extensible API to write your own Binder.

    +Spring Cloud Stream also includes a test binder for integration testing of your applications as spring-cloud-stream application. See Testing section for more details.

    Spring Cloud Stream uses Spring Boot for configuration, and the Binder abstraction makes it possible for a Spring Cloud Stream application to be flexible in how it connects to middleware. @@ -645,7 +676,7 @@ For more complex use cases, you can also package multiple binders with your appl

    Communication between applications follows a publish-subscribe model, where data is broadcast through shared topics. This can be seen in the following figure, which shows a typical deployment for a set of interacting Spring Cloud Stream applications.

    -
    +
    SCSt sensors
    @@ -679,7 +710,7 @@ When doing so, different instances of an application are placed in a competing c Each consumer binding can use the spring.cloud.stream.bindings.<channelName>.group property to specify a group name. For the consumers shown in the following figure, this property would be set as spring.cloud.stream.bindings.<channelName>.group=hdfsWrite or spring.cloud.stream.bindings.<channelName>.group=average.

    -
    +
    SCSt groups
    @@ -750,7 +781,7 @@ One or more producer application instances send data to multiple consumer applic

    Spring Cloud Stream provides a common abstraction for implementing partitioned processing use cases in a uniform fashion. Partitioning can thus be used whether the broker itself is naturally partitioned (for example, Kafka) or not (for example, RabbitMQ).

    -
    +
    SCSt partitioning
    @@ -787,14 +818,14 @@ To set up a partitioned processing scenario, you must configure both the data-pr

    Destination Binders: Components responsible to provide integration with the external messaging systems.

  • -

    Destination Bindings: Bridge between the external messaging systems and application provided Producers and Consumers of messages (created by the Destination Binders).

    +

    Bindings: Bridge between the external messaging systems and application provided Producers and Consumers of messages (created by the Destination Binders).

  • Message: The canonical data structure used by producers and consumers to communicate with Destination Binders (and thus other applications via external messaging systems).

    • -
    +
    SCSt overview
    @@ -809,17 +840,17 @@ invocation of the user code, and more.

    Binders handle a lot of the boiler plate responsibilities that would otherwise fall on your shoulders. However, to accomplish that, the binder still needs -some help in the form of minimalistic yet required set of instructions from the user, which typically come in the form of some type of configuration.

    +some help in the form of minimalistic yet required set of instructions from the user, which typically come in the form of some type of binding configuration.

    While it is out of scope of this section to discuss all of the available binder and binding configuration options (the rest of the manual covers them extensively), -Destination Binding does require special attention. The next section discusses it in detail.

    +Binding as a concept, does require special attention. The next section discusses it in detail.

    -

    Destination Bindings

    +

    Bindings

    -

    As stated earlier, Destination Bindings provide a bridge between the external messaging system and application-provided Producers and Consumers.

    +

    As stated earlier, Bindings provide a bridge between the external messaging system (e.g., queue, topic etc.) and application-provided Producers and Consumers.

    The following example shows a fully configured and functioning Spring Cloud Stream application that receives the payload of the message @@ -848,21 +879,110 @@ public class SampleApplication {

    Unlike previous versions of spring-cloud-stream which relied on @EnableBinding and @StreamListener annotations, the above example looks no different then any vanilla spring-boot application. It defines a single bean of type Function and that it is. So, how does it became spring-cloud-stream application? -It became spring-cloud-stream application simply based on the presence of spring-cloud-stream and binder dependencies -and auto-configuration classes on the classpath, which by default look for beans of type Supplier, Function or Consumer -to bind to destinations exposed by the provided binder following certain naming conventions and -rules to avoid extra configuration. -More details are in the Spring Cloud Function support section, but to finish making sense of the above sample; -Assuming that spring-cloud-stream and binder dependencies are on the classpath, a single bean of type Function defined -in the above configuration' is treated as message handler and is bound to "input" and "output" binding -destinations the identical way as you would explicitly do with @StreamListener in the previous versions of spring-cloud-stream.

    +It becomes spring-cloud-stream application simply based on the presence of spring-cloud-stream and binder dependencies +and auto-configuration classes on the classpath effectively setting the context for your boot application as spring-cloud-stream application. +And in this context beans of type Supplier, Function or Consumer are treated as defacto message handlers triggering +binding of to destinations exposed by the provided binder following certain naming conventions and +rules to avoid extra configuration.

    +
    +
    +

    Binding and Binding names

    +
    +

    Binding is an abstraction that represents a bridge between sources and targets exposed by the binder and user code, +This abstraction has a name and while we try to do our best to limit configuration required to run spring-cloud-stream applications, +being aware of such name(s) is necessary for cases where additional per-binding configuration is required.

    -

    In previous versions of spring-cloud-stream binding destinations, mentioned in previous paragraph, derived from the @EnableBinding +

    Throughout this manual you will see examples of configuration properties such as spring.cloud.stream.bindings.input.destination=myQueue. +The input segment in this property name is what we refer to as binding name and it could derive via several mechanisms. +The following sub-sections will describe the naming conventions and configuration elements used by spring-cloud-stream to control binding names.

    +
    +
    +
    Functional binding names
    +
    +

    Unlike the explicit annotation-based support (legacy) used in the previous versions of spring-cloud-stream via annotations, the functional +programming model follows a simple convention when it comes to binding names thus greatly simplifying application configuration. +Let’s look at the first example:

    +
    +
    +
    +
    @SpringBootApplication
+public class SampleApplication {
+
+	@Bean
+	public Function<String, String> uppercase() {
+	    return value -> value.toUpperCase();
+	}
+}
    +
    +
    +
    +

    In the preceding example we have an application with a single function which acts as message handler. As a Function it has an +input and output. +The naming convention used to name input and output bindings is as follows:

    +
    +
    +
      +
    • +

      input - <functionName> + -in- + <index>

      +
      • +
    • +

      output - <functionName> + -out- + <index>

      +
      • +
    +
    +
    +

    The in and out corresponds to the type of binding (such as input or output). +The index is the index of the input or output binding. It is always 0 for typical single input/output function, +so it’s only relevant for Functions with multiple input and output arguments.

    +
    +
    +

    So if for example you would want to map the input of this function to a remote +destination (e.g., topic, queue etc) called "my-topic" you would do so with the following property:

    +
    +
    +
    +
    --spring.cloud.stream.bindings.uppercase-in-0.destination=my-topic
    +
    +
    +
    +

    Note how uppercase-in-0 is used as a segment in property name. The same goes for uppercase-out-0.

    +
    +
    +

    Descriptive Binding Names

    +
    +
    +

    Some times to improve readability you may want to give your binding a more descriptive names (such as 'account', 'orders` etc). +You can do it with spring.cloud.stream.function.bindings.<binding-name> property.

    +
    +
    +

    For example,

    +
    +
    +
    +
    --spring.cloud.stream.function.bindings.uppercase-in-0=input`
    +
    +
    +
    +

    In the preceding example you mapped and effectively renamed uppercase-in-0 binding name to input. Now all configuration +properties can refer to input binding name instead (e.g., --spring.cloud.stream.bindings.input.destination=my-topic).

    +
    +
    +

    For more on properties and other configuration options please see Configuration Options section.

    +
    +
    +
    +
    Annotation-based binding names (legacy)
    +
    +

    In previous versions of spring-cloud-stream binding names and in fact implementations, derived from the @EnableBinding annotation which typically would take one or more interface classes as parameters. The parameters are referred to as bindings, and they contain methods representing bindable components.

    +

    For compliance with legacy style applications we still support this annotation-based programming model and you can get more information about it in +Annotation-based support (legacy) section (sub-section of the Programming Model section).

    +
    +

    Spring Cloud Stream already provides binding interfaces for typical message exchange contracts, which include:

    @@ -959,6 +1079,8 @@ and so on.

    In this case, an implementation of PollableMessageSource is bound to the orders “channel”. See Using Polled Consumers for more details.

    +
    +

    Producing and Consuming Messages

    @@ -1052,6 +1174,124 @@ but for all other cases it is.
    +
    Suppliers (Sources)
    +
    +

    Function and Consumer are pretty straightforward when it comes to how their invocation is triggered. They are triggered based +on data (events) sent to the destination they are bound to. In other words, they are classic event-driven components.

    +
    +
    +

    However, Supplier is in its own category when it comes to triggering. Since it is, by definition, the source (the origin) of the data, it does not +subscribe to any in-bound destination and, therefore, has to be triggered by some other mechanism(s). +There is also a question of Supplier implementation, which could be imperative or reactive and which directly relates to the triggering of such suppliers.

    +
    +
    +

    Consider the following sample:

    +
    +
    +
    +
    @SpringBootApplication
+public static class SupplierConfiguration {
+
+	@Bean
+	public Supplier<String> stringSupplier() {
+		return () -> "Hello from Supplier";
+	}
+}
    +
    +
    +
    +

    The preceding Supplier bean produces a string whenever its get() method is invoked. However, who invokes this method and how often? +The framework provides a default polling mechanism (answering the question of "Who?") that will trigger the invocation of the supplier and by default it will do so +every second (answering the question of "How often?"). +In other words, the above configuration produces a single message every second and each message is sent to an output destination that is exposed by the binder. +To learn how to customize the polling mechanism, see Polling Configuration Properties section.

    +
    +
    +

    Consider a different sample:

    +
    +
    +
    +
    @SpringBootApplication
+public static class SupplierConfiguration {
+
+	@Bean
+	public Supplier<Flux<String>> stringSupplier() {
+		return () -> Flux.from(emitter -> {
+			while (true) {
+				try {
+					emitter.onNext("Hello from Supplier");
+					Thread.sleep(1000);
+				} catch (Exception e) {
+					// ignore
+				}
+			}
+		});
+	}
+}
    +
    +
    +
    +

    The preceding Supplier bean adopts the reactive programming style. Typically, and unlike the imperative supplier, +it should be triggered only once, given that the invocation of its get() method produces (supplies) the continuous stream of messages and not an +individual message.

    +
    +
    +

    The framework recognizes the difference in the programming style and guarantees that such a supplier is triggered only once.

    +
    +
    +

    However, imagine the use case where you want to poll some data source and return a finite stream of data representing the result set. +The reactive programming style is a perfect mechanism for such a Supplier. However, given the finite nature of the produced stream, +such Supplier still needs to be invoked periodically.

    +
    +
    +

    Consider the following sample, which emulates such use case by producing a finite stream of data:

    +
    +
    +
    +
    @SpringBootApplication
+public static class SupplierConfiguration {
+
+	@PollableBean
+	public Supplier<Flux<String>> stringSupplier() {
+		return () -> Flux.just("hello", "bye");
+	}
+}
    +
    +
    +
    +

    The bean itself is annotated with PollableBean annotation (sub-set of @Bean), thus signaling to the framework that although the implementation +of such a supplier is reactive, it still needs to be polled.

    +
    +
    +
    Polling Configuration Properties
    +
    +

    The following properties are exposed by org.springframework.cloud.stream.config.DefaultPollerProperties and are prefixed with +spring.cloud.stream.poller:

    +
    +
    +
    +
    fixedDelay
    +
    +

    Fixed delay for default poller in milliseconds.

    +
    +

    Default: 1000L.

    +
    +
    +
    maxMessagesPerPoll
    +
    +

    Maximum messages for each polling event of the default poller.

    +
    +

    Default: 1L.

    +
    +
    +
    +
    +
    +

    For example --spring.cloud.stream.poller.fixed-delay=2000 sets the poller interval to poll every two seconds.

    +
    +
    +
    +
    Content-based routing with functions

    Routing with functions can be achieved by relying on RoutingFunction available in Spring Cloud Function 3.0. All you need to do is enable it via @@ -1229,27 +1469,13 @@ provided by Project Reactor - Tuples. However, we are working on allowi -While simple function binding destinations are usually named "input" and "output" (see the next section for exception to that rule), -and for the most parts are hidden from the typical user’s concerns, we can not rely on the same naming convention here. -So, this is where understanding of the naming convention for binding destinations is important. +Please refer to Binding and Binding names section to understand the naming convention used to establish binding names +used by such application.

    -

    Binding naming convention:

    -
    -
    -
      -
    • -

      input - <functionName> + .in. + <index>

      -
      • -
    • -

      output - <functionName> + .out. + <index>

      -
      • -
    -
    -

    Let’s look at the few samples:

    @@ -1273,16 +1499,16 @@ public class SampleApplication { and produces a single output of type String.

    -

    So, for the above example the two input bindings will be gather.in.0 and gather.in.1 and for consistency the -output binding also follows the same convention and is named gather.out.0.

    +

    So, for the above example the two input bindings will be gather_in_0 and gather-in-1 and for consistency the +output binding also follows the same convention and is named gather_out_0.

    Knowing that will allow you to set binding specific properties the same way you did with @StreamListener. -For example, the following will override content-type for gather.in.0 binding:

    +For example, the following will override content-type for gather-in-0 binding:

    -
    --spring.cloud.stream.bindings.gather.in.0.content-type=text/plain
    +
    --spring.cloud.stream.bindings.gather-in-0.content-type=text/plain
    @@ -1310,8 +1536,8 @@ public class SampleApplication { takes single input of type Integer and produces two outputs (both of type String).

    -

    So, for the above example the input binding is gather.in.0 and the -output bindings are gather.out.0 and gather.out.1.

    +

    So, for the above example the input binding is gather-in-0 and the +output bindings are gather-out-0 and gather-out-1.

    And you test it with the following code:

    @@ -1374,24 +1600,18 @@ So first, as mentioned before, we need to notice that there is a a conflict (mor we need to resolve it by providing spring.cloud.function.definition property pointing to the actual function we want to bind. Except here we will use ; delimiter to point to both functions (see test case below).

    -
    -

    As with functions with multiple inputs/outputs we can no longer rely on the naming convention for -destination bindings used by functions with single inputs/outputs. So we follow the same convention as -for functions with multiple inputs/outputs:

    -
    -
    -
      -
    • -

      input - <functionName> + .in. + <index>

      -
      • -
    • -

      output - <functionName> + .out. + <index>

      -
      • -
    -
    -
    -

    This means that the above configuration will result in the following destination bindings: -uppercase.in.0, uppercase.out.0, reverse.in.0 and reverse.out.0.

    +
    + + + + + +
    + + +As with functions with multiple inputs/outputs, please refer to Binding and Binding names section to understand the naming +convention used to establish binding names used by such application. +

    And you test it with the following code:

    @@ -1437,9 +1657,69 @@ public Function<List<Person>, Person> findFirstPerson() {
    +
    +
    Spring Integration flow as functions
    +
    +

    When you implement a function, you may have complex requirements that fit the category +of Enterprise Integration Patterns (EIP). These are best handled by using a +framework such as Spring Integration (SI), which is a reference implementation of EIP.

    +
    +
    +

    Thankfully SI already provides support for exposing integration flows as functions via +Integration flow as gateway +Consider the following sample:

    +
    +
    +
    +
    @SpringBootApplication
+public class FunctionSampleSpringIntegrationApplication {
+
+	public static void main(String[] args) {
+		SpringApplication.run(FunctionSampleSpringIntegrationApplication.class, args);
+	}
+
+	@Bean
+	public IntegrationFlow uppercaseFlow() {
+		return IntegrationFlows.from(MessageFunction.class, "uppercase")
+				.<String, String>transform(String::toUpperCase)
+				.logAndReply(LoggingHandler.Level.WARN);
+	}
+
+	public interface MessageFunction extends Function<Message<String>, Message<String>> {
+
+	}
+}
    +
    +
    +
    +

    For those who are familiar with SI you can see we define a bean of type IntegrationFlow where we +declare an integration flow that we want to expose as a Function<String, String> (using SI DSL) called uppercase. +The MessageFunction interface lets us explicitly declare the type of the inputs and outputs for proper type conversion. +See Content Type Negotiation section for more on type conversion.

    +
    +
    +

    To receive raw input you can use from(Function.class, …​).

    +
    +
    +

    The resulting function is bound to the input and output destinations exposed by the target binder.

    +
    +
    + + + + + +
    + + +Please refer to Binding and Binding names section to understand the naming +convention used to establish binding names used by such application. +
    +
    +
    -

    Annotation-based support

    +

    Annotation-based support (legacy)

    As mentioned earlier you can also use Spring Integration annotations based configuration or Spring Cloud Stream annotation based configuration.

    @@ -1853,7 +2133,7 @@ the error back to the messaging system (re-queue, DLQ, and others).

    There are two types of application-level error handling. Errors can be handled at each binding subscription or a global handler can handle all the binding subscription errors. Let’s review the details.

    -
    +
    custom vs global error channels
    @@ -2184,7 +2464,7 @@ This section provides information about the main concepts behind the Binder SPI,

    The following image shows the general relationship of producers and consumers:

    -
    +
    producers consumers
    @@ -2795,6 +3075,13 @@ See the producer property useNativeEncoding.

    Default: false.

    +
    multiplex
    +
    +

    When set to true, the underlying binder will natively multiplex destinations on the same input binding.

    +
    +

    Default: false.

    +
    +
    @@ -2855,6 +3142,22 @@ See Partitioning Support.

    Default: null.

    +
    partitionKeyExtractorName
    +
    +

    The name of the bean that implements PartitionKeyExtractorStrategy. Used to extract a key used to compute +the partition id (see 'partitionSelector*'). Mutually exclusive with 'partitionKeyExpression'.

    +
    +

    Default: null.

    +
    +
    +
    partitionSelectorName
    +
    +

    The name of the bean that implements PartitionSelectorStrategy. Used to determine partition id based +on partition key (see 'partitionKeyExtractor*'). Mutually exclusive with 'partitionSelectorExpression'.

    +
    +

    Default: null.

    +
    +
    partitionSelectorExpression

    A SpEL expression for customizing partition selection. @@ -2913,37 +3216,38 @@ See Error Handling<

    Using Dynamically Bound Destinations

    -

    Besides the channels defined by using @EnableBinding, Spring Cloud Stream lets applications send messages to dynamically bound destinations. +

    Aside from static destinations, Spring Cloud Stream lets applications send messages to dynamically bound destinations. This is useful, for example, when the target destination needs to be determined at runtime. -Applications can do so by using the BinderAwareChannelResolver bean, registered automatically by the @EnableBinding annotation.

    +Applications can do so in one of two ways

    +
    +
    +

    BinderAwareChannelResolver

    +
    +
    +

    The BinderAwareChannelResolver is a special bean registered automatically by the framework. +You can autowire this bean into your application and use it to resolve output destination at runtime

    The 'spring.cloud.stream.dynamicDestinations' property can be used for restricting the dynamic destination names to a known set (whitelisting). If this property is not set, any destination can be bound dynamically.

    -

    The BinderAwareChannelResolver can be used directly, as shown in the following example of a REST controller using a path variable to decide the target channel:

    +

    The following example demonstrates one of the common scenarios where REST controller uses a path variable to determine target destination:

    -
    @EnableBinding
+
    @SpringBootApplication
 @Controller
 public class SourceWithDynamicDestination {
 
     @Autowired
     private BinderAwareChannelResolver resolver;
 
-    @RequestMapping(path = "/{target}", method = POST, consumes = "*/*")
-    @ResponseStatus(HttpStatus.ACCEPTED)
-    public void handleRequest(@RequestBody String body, @PathVariable("target") target,
-           @RequestHeader(HttpHeaders.CONTENT_TYPE) Object contentType) {
-        sendMessage(body, target, contentType);
-    }
-
-    private void sendMessage(String body, String target, Object contentType) {
-        resolver.resolveDestination(target).send(MessageBuilder.createMessage(body,
-                new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, contentType))));
-    }
+    @RequestMapping(value="/{target}")
+	@ResponseStatus(HttpStatus.ACCEPTED)
+	public void send(@RequestBody String body, @PathVariable("target") String target){
+		resolver.resolveDestination(target).send(new GenericMessage<String>(body));
+	}
 }
    @@ -2958,54 +3262,41 @@ curl -H "Content-Type: application/json" -X POST -d "order-1" http://localhost:8
    -

    The destinations, 'customers' and 'orders', are created in the broker (in the exchange for Rabbit or in the topic for Kafka) with names of 'customers' and 'orders', and the data is published to the appropriate destinations.

    +

    The destinations, 'customers' and 'orders', are created in the broker (in the exchange for Rabbit or in the topic for Kafka) +with names of 'customers' and 'orders', and the data is published to the appropriate destinations.

    -

    The BinderAwareChannelResolver is a general-purpose Spring Integration DestinationResolver and can be injected in other components — for example, in a router using a SpEL expression based on the target field of an incoming JSON message. The following example includes a router that reads SpEL expressions:

    +

    spring.cloud.stream.sendto.destination

    +
    +
    +

    You can also delegate to the framework to dynamically resolve the output destination by specifying spring.cloud.stream.sendto.destination header +set to the name of the destination to be resolved.

    +
    +
    +

    Consider the following example:

    -
    @EnableBinding
+
    @SpringBootApplication
 @Controller
 public class SourceWithDynamicDestination {
 
-    @Autowired
-    private BinderAwareChannelResolver resolver;
-
-
-    @RequestMapping(path = "/", method = POST, consumes = "application/json")
-    @ResponseStatus(HttpStatus.ACCEPTED)
-    public void handleRequest(@RequestBody String body, @RequestHeader(HttpHeaders.CONTENT_TYPE) Object contentType) {
-        sendMessage(body, contentType);
-    }
-
-    private void sendMessage(Object body, Object contentType) {
-        routerChannel().send(MessageBuilder.createMessage(body,
-                new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, contentType))));
-    }
-
-    @Bean(name = "routerChannel")
-    public MessageChannel routerChannel() {
-        return new DirectChannel();
-    }
-
     @Bean
-    @ServiceActivator(inputChannel = "routerChannel")
-    public ExpressionEvaluatingRouter router() {
-        ExpressionEvaluatingRouter router =
-            new ExpressionEvaluatingRouter(new SpelExpressionParser().parseExpression("payload.target"));
-        router.setDefaultOutputChannelName("default-output");
-        router.setChannelResolver(resolver);
-        return router;
-    }
+	public Function<String, Message<String>> destinationAsPayload() {
+		return value -> {
+			return MessageBuilder.withPayload(value)
+				.setHeader("spring.cloud.stream.sendto.destination", value).build();};
+	}
 }
    -

    The Router Sink Application uses this technique to create the destinations on-demand.

    +

    Albeit trivial you can clearly see in this example, our output is a Message with spring.cloud.stream.sendto.destination header +set to the value of he input argument. The framework will consult this header and will attempt to create or discover +destination with that name and send output to it.

    -

    If the channel names are known in advance, you can configure the producer properties as with any other destination. +

    If destination names are known in advance, you can configure the producer properties as with any other destination. Alternatively, if you register a NewDestinationBindingCallback<> bean, it is invoked just before the binding is created. The callback takes the generic type of the extended producer properties used by the binder. It has one method:

    @@ -3525,6 +3816,8 @@ When a binder configuration requires them, it is important to set both values co

    To begin bridging the gap between unit and integration testing we’ve developed a new test binder which uses Spring Integration framework as an in-JVM Message Broker essentially giving you the best of both worlds - a real binder without the networking.

    +
    +

    Test Binder configuration

    To enable Spring Integration Test Binder all you need is:

    @@ -3591,13 +3884,17 @@ as an in-JVM Message Broker essentially giving you the best of both worlds - a r </dependency>
    +
    +
    +

    Test Binder usage

    Now you can test your microservice as a simple unit test

    @SpringBootApplication
-public class DemoTestBinderApplication {
+public class SampleTests {
+    @Bean
 	public Function<String, String> echo() {
 		return value -> value;
 	}
@@ -3607,29 +3904,84 @@ public class DemoTestBinderApplication {
 
 @Test
 public void sampleTest() {
-	ApplicationContext context = new SpringApplicationBuilder(
-					TestChannelBinderConfiguration.class,
-					DemoTestBinderApplication.class)
-				.web(WebApplicationType.NONE).run("--spring.cloud.function.definition=echo");
-	InputDestination source = context.getBean(InputDestination.class);
-	OutputDestination target = context.getBean(OutputDestination.class);
-	source.send(new GenericMessage<byte[]>("hello".getBytes()));
-	System.out.println("Result: " + new String(target.receive().getPayload()));
+	try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						SampleTests.class))
+				.run("--spring.cloud.function.definition=echo")) {
+		InputDestination source = context.getBean(InputDestination.class);
+		OutputDestination target = context.getBean(OutputDestination.class);
+		source.send(new GenericMessage<byte[]>("hello".getBytes()));
+		assertThat(target.receive().getPayload()).isEqualTo("hello".getBytes());
+	}
 }
    -

    or with legacy annotation-based configuration

    +

    For cases where you have multiple bindings and/or multiple inputs and outputs, the send() and receive() +methods of InputDestination and OutputDestination are overriden to allow you to provide index of the input and output destination.

    +
    +
    +

    Consider the following sample:

    +
    +
    +
    +
    @EnableAutoConfiguration
+public static class SampleFunctionConfiguration {
+
+	@Bean
+	public Function<String, String> uppercase() {
+		return value -> value.toUpperCase();
+	}
+
+	@Bean
+	public Function<String, String> reverse() {
+		return value -> new StringBuilder(value).reverse().toString();
+	}
+}
    +
    +
    +
    +

    and the actual test

    +
    +
    +
    +
    @Test
+public void testMultipleFunctions() {
+	try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+			TestChannelBinderConfiguration.getCompleteConfiguration(
+					SampleFunctionConfiguration.class))
+							.run("--spring.cloud.function.definition=uppercase;reverse")) {
+		context.getBean(InputDestination.class);
+
+		InputDestination inputDestination = context.getBean(InputDestination.class);
+		OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+		Message<byte[]> inputMessage = MessageBuilder.withPayload("Hello".getBytes()).build();
+		inputDestination.send(inputMessage, 0);
+		inputDestination.send(inputMessage, 1);
+
+		Message<byte[]> outputMessage = outputDestination.receive(0, 0);
+		assertThat(outputMessage.getPayload()).isEqualTo("HELLO".getBytes());
+
+		outputMessage = outputDestination.receive(0, 1);
+		assertThat(outputMessage.getPayload()).isEqualTo("olleH".getBytes());
+	}
+}
    +
    +
    +
    +

    Note, that first we need to provide spring.cloud.function.definition property as described in Multiple functions in a single application section +to declare which functions we intend to use for binding and then use their index (the order of definition in the spring.cloud.function.definition property) +to send/receive messages.

    +
    +
    +

    You can also use this binder with legacy annotation-based configuration:

    @SpringBootApplication
 @EnableBinding(Processor.class)
-public class DemoTestBinderApplication {
-
-	public static void main(String[] args) {
-		SpringApplication.run(DemoTestBinderApplication.class, args);
-	}
+public class LegacyStreamApplication {
 
 	@StreamListener(Processor.INPUT)
 	@SendTo(Processor.OUTPUT)
@@ -3642,14 +3994,14 @@ public class DemoTestBinderApplication {
 
 @Test
 public void sampleTest() {
-	ApplicationContext context = new SpringApplicationBuilder(
-					TestChannelBinderConfiguration.class,
-					DemoTestBinderApplication.class)
-				.web(WebApplicationType.NONE).run();
-	InputDestination source = context.getBean(InputDestination.class);
-	OutputDestination target = context.getBean(OutputDestination.class);
-	source.send(new GenericMessage<byte[]>("hello".getBytes()));
-	System.out.println("Result: " + new String(target.receive().getPayload()));
+	try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						LegacyStreamApplication.class)).run()) {
+		InputDestination source = context.getBean(InputDestination.class);
+		OutputDestination target = context.getBean(OutputDestination.class);
+		source.send(new GenericMessage<byte[]>("hello".getBytes()));
+		assertThat(target.receive().getPayload()).isEqualTo("hello".getBytes());
+	}
 }
    @@ -3674,8 +4026,9 @@ light-weight microservices architectures rather then general purpose messaging a
    +
    -

    Spring Integration Test Binder and PollableMessageSource

    +

    Test Binder and PollableMessageSource

    Spring Integration Test Binder also allows you to write tests when working with PollableMessageSource (see Using Polled Consumers for more details).

    @@ -3836,141 +4189,6 @@ In this situation, you can simply remove the dependency for kafka binder from yo
    -

    Metrics Emitter

    -
    -
    -

    Spring Boot Actuator provides dependency management and auto-configuration for Micrometer, an application metrics -facade that supports numerous monitoring systems.

    -
    -
    -

    Spring Cloud Stream provides support for emitting any available micrometer-based metrics to a binding destination, allowing for periodic -collection of metric data from stream applications without relying on polling individual endpoints.

    -
    -
    -

    Metrics Emitter is activated by defining the spring.cloud.stream.bindings.applicationMetrics.destination property, -which specifies the name of the binding destination used by the current binder to publish metric messages.

    -
    -
    -

    For example:

    -
    -
    -
    -
    spring.cloud.stream.bindings.applicationMetrics.destination=myMetricDestination
    -
    -
    -
    -

    The preceding example instructs the binder to bind to myMetricDestination (that is, Rabbit exchange, Kafka topic, and others).

    -
    -
    -

    The following properties can be used for customizing the emission of metrics:

    -
    -
    -
    -
    spring.cloud.stream.metrics.key
    -
    -

    The name of the metric being emitted. Should be a unique value per application.

    -
    -

    Default: ${spring.application.name:${vcap.application.name:${spring.config.name:application}}}

    -
    -
    -
    spring.cloud.stream.metrics.properties
    -
    -

    Allows white listing application properties that are added to the metrics payload

    -
    -

    Default: null.

    -
    -
    -
    spring.cloud.stream.metrics.meter-filter
    -
    -

    Pattern to control the 'meters' one wants to capture. -For example, specifying spring.integration.* captures metric information for meters whose name starts with spring.integration.

    -
    -

    Default: all 'meters' are captured.

    -
    -
    -
    spring.cloud.stream.metrics.schedule-interval
    -
    -

    Interval to control the rate of publishing metric data.

    -
    -

    Default: 1 min

    -
    -
    -
    -
    -
    -

    Consider the following:

    -
    -
    -
    -
    java -jar time-source.jar \
-    --spring.cloud.stream.bindings.applicationMetrics.destination=someMetrics \
-    --spring.cloud.stream.metrics.properties=spring.application** \
-    --spring.cloud.stream.metrics.meter-filter=spring.integration.*
    -
    -
    -
    -

    The following example shows the payload of the data published to the binding destination as a result of the preceding command:

    -
    -
    -
    -
    {
-	"name": "application",
-	"createdTime": "2018-03-23T14:48:12.700Z",
-	"properties": {
-	},
-	"metrics": [
-		{
-			"id": {
-				"name": "spring.integration.send",
-				"tags": [
-					{
-						"key": "exception",
-						"value": "none"
-					},
-					{
-						"key": "name",
-						"value": "input"
-					},
-					{
-						"key": "result",
-						"value": "success"
-					},
-					{
-						"key": "type",
-						"value": "channel"
-					}
-				],
-				"type": "TIMER",
-				"description": "Send processing time",
-				"baseUnit": "milliseconds"
-			},
-			"timestamp": "2018-03-23T14:48:12.697Z",
-			"sum": 130.340546,
-			"count": 6,
-			"mean": 21.72342433333333,
-			"upper": 116.176299,
-			"total": 130.340546
-		}
-	]
-}
    -
    -
    -
    - - - - - -
    - - -Given that the format of the Metric message has slightly changed after migrating to Micrometer, the published message will also have -a STREAM_CLOUD_STREAM_VERSION header set to 2.x to help distinguish between Metric messages from the older versions of the Spring Cloud Stream. -
    -
    -
    -
    -

    Samples