@Test public void testGetMessagesGreaterThan() { SwallowMessage message = createMessage(); messageDAO.saveMessage(TOPIC_NAME, message); Long maxMessageId = messageDAO.getMaxMessageId(TOPIC_NAME); SwallowMessage expectedMessage1 = createMessage(); messageDAO.saveMessage(TOPIC_NAME, expectedMessage1); SwallowMessage expectedMessage2 = createMessage(); messageDAO.saveMessage(TOPIC_NAME, expectedMessage2); List<SwallowMessage> messagesGreaterThan = messageDAO.getMessagesGreaterThan(TOPIC_NAME, maxMessageId, 5); Assert.assertNotNull(messagesGreaterThan); Assert.assertEquals(2, messagesGreaterThan.size()); Assert.assertTrue(expectedMessage1.equalsWithoutMessageId(messagesGreaterThan.get(0))); Assert.assertTrue(expectedMessage2.equalsWithoutMessageId(messagesGreaterThan.get(1))); }
@Override public List<SwallowMessage> getMessagesGreaterThan(String topicName, Long messageId, int size) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); DBObject gt = BasicDBObjectBuilder.start().add("$gt", MongoUtils.longToBSONTimestamp(messageId)).get(); DBObject query = BasicDBObjectBuilder.start().add(ID, gt).get(); DBObject orderBy = BasicDBObjectBuilder.start().add(ID, Integer.valueOf(1)).get(); DBCursor cursor = collection.find(query).sort(orderBy).limit(size); List<SwallowMessage> list = new ArrayList<SwallowMessage>(); try { while (cursor.hasNext()) { DBObject result = cursor.next(); SwallowMessage swallowMessage = new SwallowMessage(); try { convert(result, swallowMessage); list.add(swallowMessage); } catch (RuntimeException e) { LOG.error("Error when convert resultset to SwallowMessage.", e); } } } finally { cursor.close(); } return list; }
MessageDAOImpl implements MessageDAO { @Override public List<SwallowMessage> getMessagesGreaterThan(String topicName, Long messageId, int size) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); DBObject gt = BasicDBObjectBuilder.start().add("$gt", MongoUtils.longToBSONTimestamp(messageId)).get(); DBObject query = BasicDBObjectBuilder.start().add(ID, gt).get(); DBObject orderBy = BasicDBObjectBuilder.start().add(ID, Integer.valueOf(1)).get(); DBCursor cursor = collection.find(query).sort(orderBy).limit(size); List<SwallowMessage> list = new ArrayList<SwallowMessage>(); try { while (cursor.hasNext()) { DBObject result = cursor.next(); SwallowMessage swallowMessage = new SwallowMessage(); try { convert(result, swallowMessage); list.add(swallowMessage); } catch (RuntimeException e) { LOG.error("Error when convert resultset to SwallowMessage.", e); } } } finally { cursor.close(); } return list; } }
MessageDAOImpl implements MessageDAO { @Override public List<SwallowMessage> getMessagesGreaterThan(String topicName, Long messageId, int size) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); DBObject gt = BasicDBObjectBuilder.start().add("$gt", MongoUtils.longToBSONTimestamp(messageId)).get(); DBObject query = BasicDBObjectBuilder.start().add(ID, gt).get(); DBObject orderBy = BasicDBObjectBuilder.start().add(ID, Integer.valueOf(1)).get(); DBCursor cursor = collection.find(query).sort(orderBy).limit(size); List<SwallowMessage> list = new ArrayList<SwallowMessage>(); try { while (cursor.hasNext()) { DBObject result = cursor.next(); SwallowMessage swallowMessage = new SwallowMessage(); try { convert(result, swallowMessage); list.add(swallowMessage); } catch (RuntimeException e) { LOG.error("Error when convert resultset to SwallowMessage.", e); } } } finally { cursor.close(); } return list; }  }
MessageDAOImpl implements MessageDAO { @Override public List<SwallowMessage> getMessagesGreaterThan(String topicName, Long messageId, int size) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); DBObject gt = BasicDBObjectBuilder.start().add("$gt", MongoUtils.longToBSONTimestamp(messageId)).get(); DBObject query = BasicDBObjectBuilder.start().add(ID, gt).get(); DBObject orderBy = BasicDBObjectBuilder.start().add(ID, Integer.valueOf(1)).get(); DBCursor cursor = collection.find(query).sort(orderBy).limit(size); List<SwallowMessage> list = new ArrayList<SwallowMessage>(); try { while (cursor.hasNext()) { DBObject result = cursor.next(); SwallowMessage swallowMessage = new SwallowMessage(); try { convert(result, swallowMessage); list.add(swallowMessage); } catch (RuntimeException e) { LOG.error("Error when convert resultset to SwallowMessage.", e); } } } finally { cursor.close(); } return list; }  void setMongoClient(MongoClient mongoClient); @Override SwallowMessage getMessage(String topicName, Long messageId); @Override Long getMaxMessageId(String topicName); @Override SwallowMessage getMaxMessage(String topicName); @Override List<SwallowMessage> getMessagesGreaterThan(String topicName, Long messageId, int size); @Override void saveMessage(String topicName, SwallowMessage message); }
MessageDAOImpl implements MessageDAO { @Override public List<SwallowMessage> getMessagesGreaterThan(String topicName, Long messageId, int size) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); DBObject gt = BasicDBObjectBuilder.start().add("$gt", MongoUtils.longToBSONTimestamp(messageId)).get(); DBObject query = BasicDBObjectBuilder.start().add(ID, gt).get(); DBObject orderBy = BasicDBObjectBuilder.start().add(ID, Integer.valueOf(1)).get(); DBCursor cursor = collection.find(query).sort(orderBy).limit(size); List<SwallowMessage> list = new ArrayList<SwallowMessage>(); try { while (cursor.hasNext()) { DBObject result = cursor.next(); SwallowMessage swallowMessage = new SwallowMessage(); try { convert(result, swallowMessage); list.add(swallowMessage); } catch (RuntimeException e) { LOG.error("Error when convert resultset to SwallowMessage.", e); } } } finally { cursor.close(); } return list; }  void setMongoClient(MongoClient mongoClient); @Override SwallowMessage getMessage(String topicName, Long messageId); @Override Long getMaxMessageId(String topicName); @Override SwallowMessage getMaxMessage(String topicName); @Override List<SwallowMessage> getMessagesGreaterThan(String topicName, Long messageId, int size); @Override void saveMessage(String topicName, SwallowMessage message); static final String ID; static final String CONTENT; static final String VERSION; static final String SHA1; static final String GENERATED_TIME; static final String PROPERTIES; static final String INTERNAL_PROPERTIES; static final String TYPE; static final String SOURCE_IP; }