Spring Cloud stream - Routing to multiple destinations streamBridge-CodePudding

I have 3 services, "service-1" which dynamically routs to 2 other services depending on the value of id passed to the payload.

but payment-processor always routs to topic with -out-0 "spring.cloud.stream.function.bindings.processor-out-0" service-3 is never getting called.

tried with spring.cloud.stream.sendto.destination which is deprecated but it still service -3 is never getting routed

service-1

    @Bean
    public Function<Flux<String>, Flux<Object>> processor() {
        return messageFlux -> messageFlux.flatMap(
                stringMessage -> {
                    try {
                        payload = jsonMapper.readValue(stringMessage, Map.class);
                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }
                    logger.info("Payment payload  :"   payload.get("id"));
                    if(payload.get("id").equalsIgnoreCase("e-payment"))
                        streamBridge.send("processor-out-0",stringMessage);
                    else
                        streamBridge.send("processor-out-1",stringMessage);
                    return Mono.just(stringMessage);
                });
    }

    @Bean
    public Consumer<String> consumer() {
        return data -> logger.info("Payment Response :"   data);
    }

service 1 application.properties

server.port = 9000
spring.application.name=payment-process
spring.cloud.function.definition=supplier;processor;consumer
spring.cloud.stream.source=processorCheck;processorEPayment
#bindings for payment processor
spring.cloud.stream.function.bindings.processor-out-0=e-payment
spring.cloud.stream.function.bindings.processor-out-1=check-payment
spring.cloud.stream.function.bindings.processor-in-0=payment-process
#bindings for response for payment processor
spring.cloud.stream.function.bindings.consumer-in-0=payment-response
#grouping
spring.cloud.stream.bindings.processor-in-0.group=check-service
spring.cloud.stream.bindings.processor-out-0.group=check-service

servie-2

@Bean
    public Function<String,String> processorCheck() {
        return
                data ->{
                    logger.info("Data received from :"   data);
                    return "Check Payment Successful";
                };
    }

service-2 application.properties

server.port = 9002
spring.application.name=check-service
spring.cloud.function.definition=processorCheck
spring.cloud.stream.source=processorCheck
spring.cloud.stream.bindings.processorCheck-in-0.destination=check-payment
spring.cloud.stream.bindings.processorCheck-out-0.destination=payment-response
spring.cloud.stream.bindings.processorCheck-in-0.group=check-service
spring.cloud.stream.bindings.processorCheck-out-0.group=check-service
spring.cloud.stream.function.routing.enabled=true

consumer service-3

    @Bean
    public Function<String,String> processorEPayment() {
        return
         data ->{
                 logger.info("Data received from :"   data);
                 return "E-Payment Successful";
         };
    }

service-3 application.properties

server.port = 9001
spring.application.name=e-payment-service
spring.cloud.function.definition=processorEPayment
#input to e-payment service topic
spring.cloud.stream.bindings.processorEPayment-in-0.destination=e-payment
spring.cloud.stream.source=processorEPayment
#output response to e-payment topic
spring.cloud.stream.bindings.processorEPayment-out-0.destination=payment-response
spring.cloud.stream.bindings.processorEPayment-in-0.group=e-payment-service
spring.cloud.stream.bindings.processorEPayment-out-0.group=e-payment-service
spring.cloud.stream.function.routing.enabled=true

CodePudding user response：

In your service-1, in the first StreamBridge send, you can do this - streamBridge.send("e-payment",stringMessage); The second StreamBridge send call becomes - streamBridge.send("check-payment",stringMessage); They will be sent to topics, e-payment and check-payment respectively.

Remove these two properties in service-1: (they are unnecessary and may change the semantics on the bindings).

spring.cloud.stream.function.bindings.processor-out-0=e-payment
spring.cloud.stream.function.bindings.processor-out-1=check-payment

You need to add a destination for the output binding.

spring.cloud.stream.bindings.processor-out-0.destination=payment-response

The consumer then will consume from this topic (since you change the binding on the consumer to reflect this topic).