@Test public void validateSuccessfullConsumeWithEmptyQueue() throws Exception { Map<String, List<String>> routingMap = new HashMap<>(); routingMap.put("key1", Arrays.asList("queue1")); Map<String, String> exchangeToRoutingKeymap = new HashMap<>(); exchangeToRoutingKeymap.put("myExchange", "key1"); Connection conn = new TestConnection(exchangeToRoutingKeymap, routingMap); conn.createChannel().basicPublish("myExchange", "key1", null, "hello Joe".getBytes()); try (AMQPConsumer consumer = new AMQPConsumer(conn, "queue1")) { GetResponse response = consumer.consume(); assertNotNull(response); } }
public GetResponse consume() { try { return this.channel.basicGet(this.queueName, true); } catch (IOException e) { logger.error("Failed to receive message from AMQP; " + this + ". Possible reasons: Queue '" + this.queueName + "' may not have been defined", e); throw new ProcessException(e); } }
AMQPConsumer extends AMQPWorker { public GetResponse consume() { try { return this.channel.basicGet(this.queueName, true); } catch (IOException e) { logger.error("Failed to receive message from AMQP; " + this + ". Possible reasons: Queue '" + this.queueName + "' may not have been defined", e); throw new ProcessException(e); } } }
AMQPConsumer extends AMQPWorker { public GetResponse consume() { try { return this.channel.basicGet(this.queueName, true); } catch (IOException e) { logger.error("Failed to receive message from AMQP; " + this + ". Possible reasons: Queue '" + this.queueName + "' may not have been defined", e); throw new ProcessException(e); } } AMQPConsumer(Connection connection, String queueName); }
AMQPConsumer extends AMQPWorker { public GetResponse consume() { try { return this.channel.basicGet(this.queueName, true); } catch (IOException e) { logger.error("Failed to receive message from AMQP; " + this + ". Possible reasons: Queue '" + this.queueName + "' may not have been defined", e); throw new ProcessException(e); } } AMQPConsumer(Connection connection, String queueName); GetResponse consume(); @Override String toString(); }
AMQPConsumer extends AMQPWorker { public GetResponse consume() { try { return this.channel.basicGet(this.queueName, true); } catch (IOException e) { logger.error("Failed to receive message from AMQP; " + this + ". Possible reasons: Queue '" + this.queueName + "' may not have been defined", e); throw new ProcessException(e); } } AMQPConsumer(Connection connection, String queueName); GetResponse consume(); @Override String toString();  }