Updated docs with routing documentation
This commit is contained in:
Oleg Zhurakousky
@@ -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<String> even() {
|
||||
return value -> {
|
||||
System.out.println("EVEN: " + value);
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Consumer<String> 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<String> even() {
|
||||
return value -> {
|
||||
System.out.println("EVEN: " + value);
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Consumer<String> 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<String>(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<String, Message<String>> 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<RabbitProducerProperties> 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 `<<spring-cloud-stream-overview-error-handling>>` 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<String>(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<String, Message<String>> 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<RabbitProducerProperties> 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
|
||||
|
||||
Reference in New Issue
Block a user