@Test public void testPoll3() throws InterruptedException { String myType = TYPE + "_"; SwallowMessage myTypeMsg = createMessage(); myTypeMsg.setType(myType); messageDAO.saveMessage(TOPIC_NAME, myTypeMsg); Set<String> messageTypeSet = new HashSet<String>(); messageTypeSet.add(myType); BlockingQueue<Message> queue = swallowBuffer.createMessageQueue(TOPIC_NAME, cid, tailMessageId, MessageFilter.createInSetMessageFilter(messageTypeSet)); Message m = queue.poll(500, TimeUnit.MILLISECONDS); while (m == null) { m = queue.poll(50, TimeUnit.MILLISECONDS); } Assert.assertEquals(myType, m.getType()); }
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);  }