@Test public void testGetMessageQueue() throws InterruptedException { Set<String> messageTypeSet = new HashSet<String>(); messageTypeSet.add(TYPE); swallowBuffer.createMessageQueue(TOPIC_NAME, cid, tailMessageId, MessageFilter.createInSetMessageFilter(messageTypeSet)); BlockingQueue<Message> queue = swallowBuffer.getMessageQueue(TOPIC_NAME, cid); Message m; while ((m = queue.poll(1, TimeUnit.SECONDS)) == null) { ; } while ((m = queue.poll(1, TimeUnit.SECONDS)) == null) { ; } Assert.assertEquals("content3", m.getContent()); }
public BlockingQueue<Message> getMessageQueue(String topicName, String cid) { return this.getTopicBuffer(topicName).getMessageQueue(cid); }
SwallowBuffer { public BlockingQueue<Message> getMessageQueue(String topicName, String cid) { return this.getTopicBuffer(topicName).getMessageQueue(cid); } }
SwallowBuffer { public BlockingQueue<Message> getMessageQueue(String topicName, String cid) { return this.getTopicBuffer(topicName).getMessageQueue(cid); }  }
SwallowBuffer { public BlockingQueue<Message> getMessageQueue(String topicName, String cid) { return this.getTopicBuffer(topicName).getMessageQueue(cid); }  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 BlockingQueue<Message> getMessageQueue(String topicName, String cid) { return this.getTopicBuffer(topicName).getMessageQueue(cid); }  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);  }