Spring Cloud Stream – Event Routing
Spring Cloud Stream Event Routing is the ability to route events to a specific event subscriber or a specific destination. Routes ‘TO’ and ‘FROM’ will be used here. Using the Spring Cloud Stream architecture, developers may create extremely scalable event-driven microservices that are integrated with common messaging platforms. With support for persistent pub/sub semantics, consumer groups, and stateful partitions, among other features, the framework offers a versatile programming paradigm that is based on well-known and established Spring idioms and best practices.
Dependencies to Spring Cloud Stream
First, we must add the appropriate binder implementation library in our application before we can activate Spring Cloud Stream. The spring-cloud-stream-binder-rabbit artifact is required as we are connecting with RabbitMQ. There are no further dependencies that need to be included because it references all other necessary libraries.
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>
Routing TO Consumer
Routing may be accomplished by using the RoutingFunction, which is provided in Spring Cloud Function 3.0. All you have to do is activate it using the application parameter –spring.cloud.stream.function.routing.enabled=true or give the spring.cloud.function.routing-expression.
Message headers
Message header means passing some message or value to the header. By setting the spring.cloud.function.routing-expression header to value “even” and “odd” will end up routing request to either odd or even functions.
Java
@SpringBootApplication public class GFGApplication { public static void main(String[] args) { // Start the Spring Boot application with Cloud Stream function routing enabled SpringApplication.run(GFGApplication. class , "--spring.cloud.stream.function.routing.enabled=true" ); } // Bean definition for handling even numbers @Bean public Consumer<String> evenConsumer() { return value -> { // Print a message indicating that the value is even System.out.println( "EVEN: " + value); }; } // Bean definition for handling odd numbers @Bean public Consumer<String> oddConsumer() { return value -> { // Print a message indicating that the value is odd System.out.println( "ODD: " + value); }; } } |
Application properties
It is possible to send the spring.cloud.function.definition or spring.cloud.function.routing-expression as application properties spring.cloud.function.routing-expression=headers[‘type’] is one example.
Java
@SpringBootApplication public class RoutingStreamApplication { public static void main(String[] args) { // Start the Spring Boot application with a routing expression for determining even or odd SpringApplication.run(RoutingStreamApplication. class , "--spring.cloud.function.routing-expression=" + "T(java.lang.System).nanoTime() % 2 == 0 ? 'even' : 'odd'" ); } // Bean definition for handling even numbers @Bean public Consumer<Integer> evenConsumer() { return value -> { // Print a message indicating that the value is even System.out.println( "EVEN: " + value); }; } // Bean definition for handling odd numbers @Bean public Consumer<Integer> oddConsumer() { return value -> { // Print a message indicating that the value is odd System.out.println( "ODD: " + value); }; } } |
Routing FROM Consumer
Spring Cloud Stream allows apps to deliver messages to dynamically bound destinations in addition to static ones. This is helpful, for instance, if runtime target destination determination is required. Applications have two options for doing this.
spring.cloud.stream.sendto.destination:
By providing the name of the destination to be resolved in the spring.cloud.stream.sendto.destination header set, you can also assign the task of dynamically resolving the output destination to the framework.
Java
@SpringBootApplication @Controller public class SourceDestination { // Define a function that sets the payload and destination header @Bean public Function<String, Message<String>> destinationAsPayload() { return value -> { // Build a message with the payload and set the destination header return MessageBuilder.withPayload(value) .setHeader( "spring.cloud.stream.sendto.destination" , value) .build(); }; } } |
Routing Sources
Finally, let us see a further common application of a route “FROM”, in which the data source is external to SCSt yet has to be sent to the correct location:
Java
@Controller public class SourceDestination { private static final String ID_KEY = "id" ; @Autowired private ObjectMapper jsonMapper; // Create an EmitterProcessor and a FluxSink to emit messages private final EmitterProcessor<Map<String, String>> processor = EmitterProcessor.create(); private final FluxSink<Map<String, String>> fluxSink = processor.sink(); // Handle HTTP POST requests with dynamic destination based on payload @PostMapping (path = "/" , consumes = "*/*" ) @ResponseStatus (HttpStatus.ACCEPTED) public void handleRequest( @RequestBody String body, @RequestHeader (HttpHeaders.CONTENT_TYPE) Object contentType) { try { // JSON body into Map Map<String, String> payload = jsonMapper.readValue(body, new TypeReference<Map<String, String>>() {}); String destination = payload.get(ID_KEY); if (destination != null ) { // Create a message with payload and set the destination header Message<Map<String, String>> message = MessageBuilder .withPayload(payload) .setHeader( "spring.cloud.stream.sendto.destination" , destination) .build(); // Emit the payload to the defined destination fluxSink.next(message.getPayload()); } else { // Handle case where destination is not specified // You might want to log or perform some other action here } } catch (Exception e) { // Handle exceptions that might occur during JSON parsing or message processing // You might want to log or perform some other action here } } // Define a Supplier bean to act as the source of the Flux @Bean public Supplier<Flux<Map<String, String>>> source() { return () -> processor; } } |
Conclusion
So, this is Spring Cloud Stream – Event Routing. With support for persistent pub/sub semantics, consumer groups, and stateful partitions, among other features, the framework offers a versatile programming paradigm that is based on well-known and established Spring idioms and best practices.
Contact Us