@Test public void testConsume() throws Exception { final Message[] messages = new Message[]{ createMessage("you", "Hello!"), createMessage("me", "Hi!"), Message.POISON_PILL, createMessage("late_for_the_party", "Hello? Anyone here?"), }; final MessageQueue queue = new SimpleMessageQueue(messages.length); for (final Message message : messages) { queue.put(message); } new Consumer("NSA", queue).consume(); assertTrue(appender.logContains("Message [Hello!] from [you] received by [NSA]")); assertTrue(appender.logContains("Message [Hi!] from [me] received by [NSA]")); assertTrue(appender.logContains("Consumer NSA receive request to terminate.")); }
public void consume() { while (true) { Message msg; try { msg = queue.take(); if (Message.POISON_PILL.equals(msg)) { LOGGER.info("Consumer {} receive request to terminate.", name); break; } } catch (InterruptedException e) { LOGGER.error("Exception caught.", e); return; } String sender = msg.getHeader(Headers.SENDER); String body = msg.getBody(); LOGGER.info("Message [{}] from [{}] received by [{}]", body, sender, name); } }
Consumer { public void consume() { while (true) { Message msg; try { msg = queue.take(); if (Message.POISON_PILL.equals(msg)) { LOGGER.info("Consumer {} receive request to terminate.", name); break; } } catch (InterruptedException e) { LOGGER.error("Exception caught.", e); return; } String sender = msg.getHeader(Headers.SENDER); String body = msg.getBody(); LOGGER.info("Message [{}] from [{}] received by [{}]", body, sender, name); } } }
Consumer { public void consume() { while (true) { Message msg; try { msg = queue.take(); if (Message.POISON_PILL.equals(msg)) { LOGGER.info("Consumer {} receive request to terminate.", name); break; } } catch (InterruptedException e) { LOGGER.error("Exception caught.", e); return; } String sender = msg.getHeader(Headers.SENDER); String body = msg.getBody(); LOGGER.info("Message [{}] from [{}] received by [{}]", body, sender, name); } } Consumer(String name, MqSubscribePoint queue); }
Consumer { public void consume() { while (true) { Message msg; try { msg = queue.take(); if (Message.POISON_PILL.equals(msg)) { LOGGER.info("Consumer {} receive request to terminate.", name); break; } } catch (InterruptedException e) { LOGGER.error("Exception caught.", e); return; } String sender = msg.getHeader(Headers.SENDER); String body = msg.getBody(); LOGGER.info("Message [{}] from [{}] received by [{}]", body, sender, name); } } Consumer(String name, MqSubscribePoint queue); void consume(); }
Consumer { public void consume() { while (true) { Message msg; try { msg = queue.take(); if (Message.POISON_PILL.equals(msg)) { LOGGER.info("Consumer {} receive request to terminate.", name); break; } } catch (InterruptedException e) { LOGGER.error("Exception caught.", e); return; } String sender = msg.getHeader(Headers.SENDER); String body = msg.getBody(); LOGGER.info("Message [{}] from [{}] received by [{}]", body, sender, name); } } Consumer(String name, MqSubscribePoint queue); void consume();  }