title: Bridge RabbitMQ listener to Flux
body: <p>I have a Reactive Spring Boot application consuming messages from RabbitMQ and persisting them in a (MongoDB) repository:</p>
  
  <pre><code>@RabbitListener(...)
  public void processMessage(Message message) {
      repository.persist(message).subscribe();
  }
  </code></pre>
  
  <p>Assuming multiple messages arriving in a short period of time, this code could deplete a configured ConnectionPool to the database. If I would receive the messages in a <code>Flux</code>, I could <code>concatMap()</code> them into the db or insert them in buckets of n documents. </p>
  
  <p>That's why I tried to implement a bridge of the given RabbitMQ listener to a self-managed Flux:</p>
  
  <pre><code>@Component
  public class QueueListenerController {
  
      private final MyMongoRepository repository;
      private final FluxProcessor&lt;Message, Message&gt; fluxProcessor;
      private final FluxSink&lt;Message&gt; fluxSink;
  
      public QueueListenerController(MyMongoRepository repository) {
          this.repository = repository;
          this.fluxProcessor = DirectProcessor.&lt;Message&gt;create().serialize();
          this.fluxSink = fluxProcessor.sink();
      }
  
      @PostConstruct
      private void postConstruct() {
          fluxProcessor.concatMap(repository::persist)
                  .subscribe();
      }
  
      @RabbitListener(bindings = @QueueBinding(
              value = @Queue(value = "my-queue", durable = "true", autoDelete = "false"),
              exchange = @Exchange(value = "amq.direct", durable = "true", autoDelete = "false")
      ))
      public void processMessage(Message message) {
          fluxSink.next(message);
      }
  }
  </code></pre>
  
  <p>This works locally and for a certain period of time but after some time (I expect 12-24 hours) it stops storing messages in the database, so I'm quite sure I'm doing something wrong.</p>
  
  <p>What would be the correct way of transforming incoming RabbitMQ messages into a <code>Flux</code> of messages? </p>