@Test public void testCreateMessageQueue1() throws InterruptedException { BlockingQueue<Message> queue = swallowBuffer.createMessageQueue(TOPIC_NAME, cid, tailMessageId); Message m; while ((m = queue.poll(1, TimeUnit.SECONDS)) == null) { ; } Assert.assertEquals("content2", m.getContent()); }
public CloseableBlockingQueue<Message> createMessageQueue(String topicName, String cid, Long tailMessageId, MessageFilter messageFilter) { return this.getTopicBuffer(topicName).createMessageQueue(cid, tailMessageId, messageFilter); }
SwallowBuffer { public CloseableBlockingQueue<Message> createMessageQueue(String topicName, String cid, Long tailMessageId, MessageFilter messageFilter) { return this.getTopicBuffer(topicName).createMessageQueue(cid, tailMessageId, messageFilter); } }
SwallowBuffer { public CloseableBlockingQueue<Message> createMessageQueue(String topicName, String cid, Long tailMessageId, MessageFilter messageFilter) { return this.getTopicBuffer(topicName).createMessageQueue(cid, tailMessageId, messageFilter); }  }
SwallowBuffer { public CloseableBlockingQueue<Message> createMessageQueue(String topicName, String cid, Long tailMessageId, MessageFilter messageFilter) { return this.getTopicBuffer(topicName).createMessageQueue(cid, tailMessageId, messageFilter); }  BlockingQueue<Message> getMessageQueue(String topicName, String cid); CloseableBlockingQueue<Message> createMessageQueue(String topicName, String cid, Long tailMessageId,
                                                             MessageFilter messageFilter); BlockingQueue<Message> createMessageQueue(String topicName, String cid, Long tailMessageId); void setCapacity(int capacity); void setThreshold(int threshold); void setDelayBase(int delayBase); void setDelayUpperbound(int delayUpperbound); void setMessageRetriever(MessageRetriever messageRetriever); }
SwallowBuffer { public CloseableBlockingQueue<Message> createMessageQueue(String topicName, String cid, Long tailMessageId, MessageFilter messageFilter) { return this.getTopicBuffer(topicName).createMessageQueue(cid, tailMessageId, messageFilter); }  BlockingQueue<Message> getMessageQueue(String topicName, String cid); CloseableBlockingQueue<Message> createMessageQueue(String topicName, String cid, Long tailMessageId,
                                                             MessageFilter messageFilter); BlockingQueue<Message> createMessageQueue(String topicName, String cid, Long tailMessageId); void setCapacity(int capacity); void setThreshold(int threshold); void setDelayBase(int delayBase); void setDelayUpperbound(int delayUpperbound); void setMessageRetriever(MessageRetriever messageRetriever);  }