PublishConfirm Operator does not respect backpressure

[smaldini]

Currently there is https://github.com/reactor/reactor-rabbitmq/blob/34629edb1cbd532587da976c4d88062b905933c1/src/main/java/reactor/rabbitmq/Sender.java#L446 and https://github.com/reactor/reactor-rabbitmq/blob/34629edb1cbd532587da976c4d88062b905933c1/src/main/java/reactor/rabbitmq/Sender.java#L457 possibly sending more onNext than expected to the downstream subscriber. Unless we configure the flatMapMany to accept an unbounded number of messages via the second prefetch argument.

[pmackowski]

Unfortunately, not only subscriber can get more messages but also more messages might be sent to RabbitMq. Looks like a race condition. For the below snippet, ~80-100 messages are sent to the broker and ~15 are confirmed.

int messages = 10000;
int confirmed = 10
Flux<OutboundMessage> msgFlux = Flux.range(0, messages).map(i -> new OutboundMessage("", queue, "".getBytes()));
sender.sendWithPublishConfirms(msgFlux)
           .take(confirmed)
           .subscribe();

[acogoluegnes]

Fixed by #67.