Publisher confirms and returns

Key points

• confirm.select switches a channel into confirm mode, and the broker then acks each publish.
• An ack for a persistent message means it reached disk, or was committed on a majority of quorum replicas.
• A basic.nack from the broker means it lost the message - resend or alert, never ignore it.
• Confirms are asynchronous, and blocking on each one costs a round trip that destroys throughput.
• Unroutable messages are dropped silently unless you publish with the mandatory flag set.
• An alternate-exchange argument catches unroutable messages without any return handler.

Example

ch.confirmSelect();                       // channel enters confirm mode

// unroutable messages come back here, but only with mandatory=true
ch.addReturnListener(ret ->
    log.error("unroutable rk={} reply={}",
              ret.getRoutingKey(), ret.getReplyText()));

// The multiple flag MUST be honoured. Confirms can arrive out of order, so
// treating every ack as cumulative drops tracking for tags never confirmed,
// and treating every nack as cumulative resends messages that were fine.
ch.addConfirmListener(
    (tag, multiple) -> {
        if (multiple) outstanding.headMap(tag, true).clear();
        else          outstanding.remove(tag);
    },
    (tag, multiple) -> {
        if (multiple) resend(outstanding.headMap(tag, true));
        else          resend(tag, outstanding.remove(tag));
    });

long seq = ch.getNextPublishSeqNo();      // the tag this publish will get
outstanding.put(seq, body);
ch.basicPublish("orders", "order.created",
                true,                     // mandatory
                MessageProperties.PERSISTENT_TEXT_PLAIN, body);

// batch instead of blocking per message:
ch.waitForConfirmsOrDie(5_000);           // once per few hundred publishes

basicPublish returning normally means nothing - confirms prove storage, mandatory proves routing.