AMQPObservableQueue implements ObservableQueue { @Override public List<String> ack(List<Message> messages) { final List<String> processedDeliveryTags = new ArrayList<>(); for (final Message message : messages) { try { logger.info("ACK message with delivery tag {}", message.getReceipt()); getOrCreateChannel().basicAck(Long.valueOf(message.getReceipt()), false); processedDeliveryTags.add(message.getReceipt()); } catch (final IOException e) { logger.error("Cannot ACK message with delivery tag {}", message.getReceipt(), e); } } return processedDeliveryTags; } AMQPObservableQueue(final ConnectionFactory factory, final Address[] addresses, final boolean useExchange,
			final AMQPSettings settings, final int batchSize, final int pollTimeInMS); @Override Observable<Message> observe(); @Override String getType(); @Override String getName(); @Override String getURI(); int getBatchSize(); AMQPSettings getSettings(); Address[] getAddresses(); @Override List<String> ack(List<Message> messages); @Override void publish(List<Message> messages); @Override void setUnackTimeout(Message message, long unackTimeout); @Override long size(); @Override void close(); int getPollTimeInMS(); void setPollTimeInMS(int pollTimeInMS);  }
@Test public void testAck() throws IOException, TimeoutException { Channel channel = mockBaseChannel(); Connection connection = mockGoodConnection(channel); final Random random = new Random(); final String name = RandomStringUtils.randomAlphabetic(30), type = "topic", routingKey = RandomStringUtils.randomAlphabetic(30); final AMQPSettings settings = new AMQPSettings(configuration).fromURI("amqp_exchange:" + name + "?exchangeType=" + type + "&routingKey=" + routingKey); AMQPObservableQueue observableQueue = new AMQPObservableQueue(mockConnectionFactory(connection), addresses, true, settings, batchSize, pollTimeMs); List<Message> messages = new LinkedList<>(); Message msg = new Message(); msg.setId("0e3eef8f-ebb1-4244-9665-759ab5bdf433"); msg.setPayload("Payload"); msg.setReceipt("1"); messages.add(msg); List<String> deliveredTags = observableQueue.ack(messages); assertNotNull(deliveredTags); }