Replacement for DirectProcessor

The issue is in reference to this question.

As per the comments provided, I tried to use StreamBridge to publish the events to the Kafka topic. Let me explain in detail by putting the code snippet before and after using Streambridge bean.

Before

1. Supplier bean & its dependency to publish the events

@Autowired
private DirectProcessor<RequestDTO> source;

@Bean
public Supplier<Flux<RequestDTO>> supplier(){
    return () -> Flux.from(source);
};

2. Bean configuration class

@Bean
public DirectProcessor<RequestDTO> publisher(){
     return DirectProcessor.create();
}

@Bean
public FluxSink<OrchestratorRequestDTO> sink(DirectProcessor<RequestDTO> publisher){
    return publisher.sink();
}

3. Class where this FluxSink bean is injected to publish the events.

@Autowired
private FluxSink<RequestDTO> sink;

public Order createOrder(RequestDTO requestDTO){
    this.sink.next(this.getOrderRequestDTO(requestDTO));
        
     // rest of the logic to return the response

}

After using StreamBridge

1. & 2. In my opinion, these steps are not required. So, removed configuring the beans of type DirectProcessor & FluxSink
2. A class where StreamBridge bean is now getting used. The binding name supplier-out-0 is configured in the application.yml

@Autowired
private StreamBridge bridge;

public Order createOrder(OrderRequestDTO orderRequestDTO){
    this.bridge.send("supplier-out-0",
                Flux.just(this.getOrderRequestDTO(requestDTO));
    // rest of the logic to return the response
}

4. This resulted in the below exception. Based on this exception, it looks like the StreamBridge bean's send() method is expecting an object of type Message<T>.

java.lang.ClassCastException: class reactor.core.publisher.FluxMapFuseable cannot be cast to class org.springframework.messaging.Message (reactor.core.publisher.FluxMapFuseable and org.springframework.messaging.Message are in unnamed module of loader 'app')
	at org.springframework.cloud.stream.function.StreamBridge.send(StreamBridge.java:143) ~[spring-cloud-stream-3.0.6.RELEASE.jar:3.0.6.RELEASE]
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
	|_ checkpoint ⇢ org.springframework.boot.actuate.metrics.web.reactive.server.MetricsWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ HTTP POST "/order/create" [ExceptionHandlingWebHandler]
Stack trace:
		at org.springframework.cloud.stream.function.StreamBridge.send(StreamBridge.java:143) ~[spring-cloud-stream-3.0.6.RELEASE.jar:3.0.6.RELEASE]
		at org.springframework.cloud.stream.function.StreamBridge.send(StreamBridge.java:116) ~[spring-cloud-stream-3.0.6.RELEASE.jar:3.0.6.RELEASE]

The event published on this Kafka topic is processed by the processor mentioned below:

@Bean
public Function<Flux<RequestDTO>, Flux<ResponseDTO>> processor(){
        return flux -> flux
                .flatMap(dto -> this.orchestratorService.orderProduct(dto))
                .doOnNext(...); 
}

If the publisher starts using the StreamBridge then what are the expected changes in the processor() method mentioned above?

Additional details:

1. Spring boot version- 2.3.12.RELEASE
2. Spring cloud version- Hoxton.SR6

Start with the StreamBridge.send call, the supplier-out-0 binding in application.yml, and the processor() function shown in the issue. Read the relevant Spring Cloud Stream and StreamBridge behavior for the stated Spring Boot and Cloud versions; done means the replacement flow and required processor changes are established without the ClassCastException.