@Test public void testUnconsumedMessagesNacked() throws TimeoutException, IOException { final Map<String, List<String>> routingMap = Collections.singletonMap("key1", Arrays.asList("queue1", "queue2")); final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1"); final TestConnection connection = new TestConnection(exchangeToRoutingKeymap, routingMap); final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true); consumer.getChannel().basicPublish("myExchange", "key1", new BasicProperties(), new byte[0]); consumer.close(); assertTrue(((TestChannel) consumer.getChannel()).isNack(0)); }
@Override public void close() throws TimeoutException, IOException { closed = true; GetResponse lastMessage = null; GetResponse response; while ((response = responseQueue.poll()) != null) { lastMessage = response; } if (lastMessage != null) { getChannel().basicNack(lastMessage.getEnvelope().getDeliveryTag(), true, true); } }
AMQPConsumer extends AMQPWorker { @Override public void close() throws TimeoutException, IOException { closed = true; GetResponse lastMessage = null; GetResponse response; while ((response = responseQueue.poll()) != null) { lastMessage = response; } if (lastMessage != null) { getChannel().basicNack(lastMessage.getEnvelope().getDeliveryTag(), true, true); } } }
AMQPConsumer extends AMQPWorker { @Override public void close() throws TimeoutException, IOException { closed = true; GetResponse lastMessage = null; GetResponse response; while ((response = responseQueue.poll()) != null) { lastMessage = response; } if (lastMessage != null) { getChannel().basicNack(lastMessage.getEnvelope().getDeliveryTag(), true, true); } } AMQPConsumer(final Connection connection, final String queueName, final boolean autoAcknowledge); }
AMQPConsumer extends AMQPWorker { @Override public void close() throws TimeoutException, IOException { closed = true; GetResponse lastMessage = null; GetResponse response; while ((response = responseQueue.poll()) != null) { lastMessage = response; } if (lastMessage != null) { getChannel().basicNack(lastMessage.getEnvelope().getDeliveryTag(), true, true); } } AMQPConsumer(final Connection connection, final String queueName, final boolean autoAcknowledge); GetResponse consume(); void acknowledge(final GetResponse response); @Override void close(); @Override String toString(); }
AMQPConsumer extends AMQPWorker { @Override public void close() throws TimeoutException, IOException { closed = true; GetResponse lastMessage = null; GetResponse response; while ((response = responseQueue.poll()) != null) { lastMessage = response; } if (lastMessage != null) { getChannel().basicNack(lastMessage.getEnvelope().getDeliveryTag(), true, true); } } AMQPConsumer(final Connection connection, final String queueName, final boolean autoAcknowledge); GetResponse consume(); void acknowledge(final GetResponse response); @Override void close(); @Override String toString();  }