Feature Requests: Reactive StreamBridge

Hi,

The upcoming Spring Cloud Stream 4.0.0 contains the new spring-cloud-stream-binder-kafka-reactive binder which works great for consuming messages - personally this is one of the flag ship features :-)

The issue is sending messages via StreamBridge as StreamBridge is not reactive: public boolean send(String bindingName, Object data) . This means, when combined with spring cloud function AWS, the message is never sent:

public class MessageSenderFunction implements Function<Flux<Map<String, String>>, Flux<String>> {

  private StreamBridge streamBridge;

  @Override
  public Flux<String> apply(Flux<Map<String, String>> flux) {
    return flux.flatMap(v -> {
      Object kafkaEvent = ...
      if (streamBridge.send("test-out-0", kafkaEvent)) { // message never sent!!
        return Mono.just("OK");
      } else {
        return Mono.error(new RuntimeException("event publishing failed"));
      }
    });
  }
}

There are two current ways around this:

1. Don't use the reactive kafka binder

2. Add a delay when using StreamBridge

      if (streamBridge.send("test-out-0", kafkaEvent)) {
        log.info("Message sent to binding = {}", kafkaEvent);
        return Mono.just("OK").delayElement(Duration.ofSeconds(1));
      }

Both are poor options.

Would it be possible to add a reactive StreamBridge or the ability to send messages where polling (i.e. @PollableBean) is not an option? Reactor kafka has a KafkaSender class which has implemented this as:

<T> Flux<SenderResult<T>> send(Publisher<? extends SenderRecord<K, V, T>> records);

Could we have something similar to the above or a way to get a handle on the above in Spring Cloud Streams?

Thanks