ConfirmedPublisher extends DiscretePublisher { @Override public void publish(Message message, DeliveryOptions deliveryOptions) throws IOException { for (int attempt = 1; attempt <= DEFAULT_RETRY_ATTEMPTS; attempt++) { if (attempt > 1) { LOGGER.info("Attempt {} to send message", attempt); } try { Channel channel = provideChannel(); message.publish(channel, deliveryOptions); LOGGER.info("Waiting for publisher ack"); channel.waitForConfirmsOrDie(); LOGGER.info("Received publisher ack"); return; } catch (IOException e) { handleIoException(attempt, e); } catch (InterruptedException e) { LOGGER.warn("Publishing message interrupted while waiting for producer ack", e); return; } } } ConfirmedPublisher(ConnectionFactory connectionFactory); @Override void publish(Message message, DeliveryOptions deliveryOptions); @Override void publish(List<Message> messages, DeliveryOptions deliveryOptions);  }
@Test public void shouldPublishMessageAndWaitForAck() throws Exception { Message message = new Message().exchange(TEST_EXCHANGE).routingKey(TEST_ROUTING_KEY); ConfirmedPublisher publisher = new ConfirmedPublisher(connectionFactory); mockConnectionOperations(); expect(channel.confirmSelect()).andReturn(null); channel.basicPublish(TEST_EXCHANGE, TEST_ROUTING_KEY, false, false, message.getBasicProperties(), message.getBodyContent()); expectLastCall().once(); channel.waitForConfirmsOrDie(); expectLastCall().once(); PowerMock.replayAll(); publisher.publish(message); PowerMock.verifyAll(); }