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@Controllerpublic 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
@Controllerpublic 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; } } |
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>
Contact Us