diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 11de62aaa..3e6f22d0f 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -568,57 +568,6 @@ Default: 1L. For example `--spring.cloud.stream.poller.fixed-delay=2000` sets the poller interval to poll every two seconds. -===== 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 -`--spring.cloud.stream.function.routing.enabled=true` application property or provide `spring.cloud.function.routing-expression` property. -Once enabled `RoutingFunction` will be bound to input destination -receiving all the messages and route them to other functions based on the provided instruction. - -Instruction could be provided with individual messages as well as application properties. - -Here are couple of samples: - -***Using message headers*** -[source,java] ----- -@SpringBootApplication -public class SampleApplication { - - public static void main(String[] args) { - SpringApplication.run(SampleApplication.class, - "--spring.cloud.stream.function.routing.enabled=true"); - } - - @Bean - public Consumer even() { - return value -> { - System.out.println("EVEN: " + value); - }; - } - - @Bean - public Consumer odd() { - return value -> { - System.out.println("ODD: " + value); - }; - } -} ----- -By default `RoutingFunction` will look for `spring.cloud.function.definition` header and if it is found its value will be treated as routing instruction. -So in the above case the value of such header should be either `odd` or `even` (the name of the function beans) to route request to available functions. - -You can also use SpEL for more dynamic scenarios via `spring.cloud.function.routing-expression` header. -For example, -setting `spring.cloud.function.routing-expression` header to value `T(java.lang.System).currentTimeMillis() % 2 == 0 ? 'even' : 'odd'` will end up semi-randomly routing request to either `odd` or `even` functions. -Also, for SpEL, the _root object_ of the evaluation context is `Message` so you can do evaluation on individual headers (or message) as well `....routing-expression=headers['type']` - -***Using application properties*** - -The `spring.cloud.function.routing-expression` and/or `spring.cloud.function.definition` -can be passed as application properties (e.g., `spring.cloud.function.routing-expression=headers['type']`. - -Passing instructions via application properties is especially important for reactive functions since given that fact that reactive -function is only invoked once to pass the Publisher, so access to the individual items is limited. ===== Reactive Functions support @@ -1199,6 +1148,164 @@ If the service activator throws a `RequeueCurrentMessageException`, the message If the listener throws a `RequeueCurrentMessageException` directly, the message will be requeued, as discussed above, and will not be sent to the error channels. +=== Event Routing + +Event Routing, in the context of Spring Cloud Stream, is the ability to either +_a) route evens to a particular even subscriber_ or +_b) route event produced by an event subscriber to a particular destination_. +Here we'll refer to it as route ‘TO’ and route ‘FROM’. + +==== Routing TO Consumer +Routing can be achieved by relying on `RoutingFunction` available in Spring Cloud Function 3.0. All you need to do is enable it via +`--spring.cloud.stream.function.routing.enabled=true` application property or provide `spring.cloud.function.routing-expression` property. +Once enabled `RoutingFunction` will be bound to input destination +receiving all the messages and route them to other functions based on the provided instruction. + +Instruction could be provided with individual messages as well as application properties. + +Here are couple of samples: + +***Using message headers*** +[source,java] +---- +@SpringBootApplication +public class SampleApplication { + + public static void main(String[] args) { + SpringApplication.run(SampleApplication.class, + "--spring.cloud.stream.function.routing.enabled=true"); + } + + @Bean + public Consumer even() { + return value -> { + System.out.println("EVEN: " + value); + }; + } + + @Bean + public Consumer odd() { + return value -> { + System.out.println("ODD: " + value); + }; + } +} +---- +By default `RoutingFunction` will look for `spring.cloud.function.definition` header and if it is found its value will be treated as routing instruction. +So in the above case the value of such header should be either `odd` or `even` (the name of the function beans) to route request to available functions. + +You can also use SpEL for more dynamic scenarios via `spring.cloud.function.routing-expression` header. +For example, +setting `spring.cloud.function.routing-expression` header to value `T(java.lang.System).currentTimeMillis() % 2 == 0 ? 'even' : 'odd'` will end up semi-randomly routing request to either `odd` or `even` functions. +Also, for SpEL, the _root object_ of the evaluation context is `Message` so you can do evaluation on individual headers (or message) as well `....routing-expression=headers['type']` + +***Using application properties*** + +The `spring.cloud.function.routing-expression` and/or `spring.cloud.function.definition` +can be passed as application properties (e.g., `spring.cloud.function.routing-expression=headers['type']`. + +Passing instructions via application properties is especially important for reactive functions since given that fact that reactive +function is only invoked once to pass the Publisher, so access to the individual items is limited. + +==== Routing FROM Consumer + +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 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 following example demonstrates one of the common scenarios where REST controller uses a path variable to determine target destination: + +[source,java] +---- +@SpringBootApplication +@Controller +public class SourceWithDynamicDestination { + + @Autowired + private BinderAwareChannelResolver resolver; + + @RequestMapping(value="/{target}") + @ResponseStatus(HttpStatus.ACCEPTED) + public void send(@RequestBody String body, @PathVariable("target") String target){ + resolver.resolveDestination(target).send(new GenericMessage(body)); + } +} +---- + +Now consider what happens when we start the application on the default port (8080) and make the following requests with CURL: + +---- +curl -H "Content-Type: application/json" -X POST -d "customer-1" http://localhost:8080/customers + +curl -H "Content-Type: application/json" -X POST -d "order-1" http://localhost:8080/orders +---- + +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. + +***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: + +[source,java] +---- +@SpringBootApplication +@Controller +public class SourceWithDynamicDestination { + + @Bean + public Function> destinationAsPayload() { + return value -> { + return MessageBuilder.withPayload(value) + .setHeader("spring.cloud.stream.sendto.destination", value).build();}; + } +} +---- + +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 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: + +[source, java] +---- +void configure(String channelName, MessageChannel channel, ProducerProperties producerProperties, + T extendedProducerProperties); +---- + +The following example shows how to use the RabbitMQ binder: + +[source, java] +---- +@Bean +public NewDestinationBindingCallback dynamicConfigurer() { + return (name, channel, props, extended) -> { + props.setRequiredGroups("bindThisQueue"); + extended.setQueueNameGroupOnly(true); + extended.setAutoBindDlq(true); + extended.setDeadLetterQueueName("myDLQ"); + }; +} +---- + +NOTE: If you need to support dynamic destinations with multiple binder types, use `Object` for the generic type and cast the `extended` argument as needed. + [[spring-cloud-stream-overview-error-handling]] === Error Handling @@ -1877,105 +1984,6 @@ See `<>` for more information. + Default: `false`. -[[dynamicdestination]] -=== Using 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 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 following example demonstrates one of the common scenarios where REST controller uses a path variable to determine target destination: - -[source,java] ----- -@SpringBootApplication -@Controller -public class SourceWithDynamicDestination { - - @Autowired - private BinderAwareChannelResolver resolver; - - @RequestMapping(value="/{target}") - @ResponseStatus(HttpStatus.ACCEPTED) - public void send(@RequestBody String body, @PathVariable("target") String target){ - resolver.resolveDestination(target).send(new GenericMessage(body)); - } -} ----- - -Now consider what happens when we start the application on the default port (8080) and make the following requests with CURL: - ----- -curl -H "Content-Type: application/json" -X POST -d "customer-1" http://localhost:8080/customers - -curl -H "Content-Type: application/json" -X POST -d "order-1" http://localhost:8080/orders ----- - -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. - -***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: - -[source,java] ----- -@SpringBootApplication -@Controller -public class SourceWithDynamicDestination { - - @Bean - public Function> destinationAsPayload() { - return value -> { - return MessageBuilder.withPayload(value) - .setHeader("spring.cloud.stream.sendto.destination", value).build();}; - } -} ----- - -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 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: - -[source, java] ----- -void configure(String channelName, MessageChannel channel, ProducerProperties producerProperties, - T extendedProducerProperties); ----- - -The following example shows how to use the RabbitMQ binder: - -[source, java] ----- -@Bean -public NewDestinationBindingCallback dynamicConfigurer() { - return (name, channel, props, extended) -> { - props.setRequiredGroups("bindThisQueue"); - extended.setQueueNameGroupOnly(true); - extended.setAutoBindDlq(true); - extended.setDeadLetterQueueName("myDLQ"); - }; -} ----- - -NOTE: If you need to support dynamic destinations with multiple binder types, use `Object` for the generic type and cast the `extended` argument as needed. [[content-type-management]] == Content Type Negotiation