@Test public void testGetMaxMessageId() { int time = (int) (System.currentTimeMillis() / 1000); int inc = 1; BSONTimestamp timestamp = new BSONTimestamp(time, inc); Long expectedMessageId = MongoUtils.BSONTimestampToLong(timestamp); ackDAO.add(TOPIC_NAME, CONSUMER_ID, MongoUtils.BSONTimestampToLong(timestamp), IP); Long maxMessageId = ackDAO.getMaxMessageId(TOPIC_NAME, CONSUMER_ID); Assert.assertEquals(expectedMessageId, maxMessageId); }
@Override public Long getMaxMessageId(String topicName, String consumerId) { DBCollection collection = this.mongoClient.getAckCollection(topicName, consumerId); DBObject fields = BasicDBObjectBuilder.start().add(MSG_ID, Integer.valueOf(1)).get(); DBObject orderBy = BasicDBObjectBuilder.start().add(MSG_ID, Integer.valueOf(-1)).get(); DBCursor cursor = collection.find(new BasicDBObject(), fields).sort(orderBy).limit(1); try { if (cursor.hasNext()) { DBObject result = cursor.next(); BSONTimestamp timestamp = (BSONTimestamp) result.get(MSG_ID); return MongoUtils.BSONTimestampToLong(timestamp); } } finally { cursor.close(); } return null; }
AckDAOImpl implements AckDAO { @Override public Long getMaxMessageId(String topicName, String consumerId) { DBCollection collection = this.mongoClient.getAckCollection(topicName, consumerId); DBObject fields = BasicDBObjectBuilder.start().add(MSG_ID, Integer.valueOf(1)).get(); DBObject orderBy = BasicDBObjectBuilder.start().add(MSG_ID, Integer.valueOf(-1)).get(); DBCursor cursor = collection.find(new BasicDBObject(), fields).sort(orderBy).limit(1); try { if (cursor.hasNext()) { DBObject result = cursor.next(); BSONTimestamp timestamp = (BSONTimestamp) result.get(MSG_ID); return MongoUtils.BSONTimestampToLong(timestamp); } } finally { cursor.close(); } return null; } }
AckDAOImpl implements AckDAO { @Override public Long getMaxMessageId(String topicName, String consumerId) { DBCollection collection = this.mongoClient.getAckCollection(topicName, consumerId); DBObject fields = BasicDBObjectBuilder.start().add(MSG_ID, Integer.valueOf(1)).get(); DBObject orderBy = BasicDBObjectBuilder.start().add(MSG_ID, Integer.valueOf(-1)).get(); DBCursor cursor = collection.find(new BasicDBObject(), fields).sort(orderBy).limit(1); try { if (cursor.hasNext()) { DBObject result = cursor.next(); BSONTimestamp timestamp = (BSONTimestamp) result.get(MSG_ID); return MongoUtils.BSONTimestampToLong(timestamp); } } finally { cursor.close(); } return null; }  }
AckDAOImpl implements AckDAO { @Override public Long getMaxMessageId(String topicName, String consumerId) { DBCollection collection = this.mongoClient.getAckCollection(topicName, consumerId); DBObject fields = BasicDBObjectBuilder.start().add(MSG_ID, Integer.valueOf(1)).get(); DBObject orderBy = BasicDBObjectBuilder.start().add(MSG_ID, Integer.valueOf(-1)).get(); DBCursor cursor = collection.find(new BasicDBObject(), fields).sort(orderBy).limit(1); try { if (cursor.hasNext()) { DBObject result = cursor.next(); BSONTimestamp timestamp = (BSONTimestamp) result.get(MSG_ID); return MongoUtils.BSONTimestampToLong(timestamp); } } finally { cursor.close(); } return null; }  void setMongoClient(MongoClient mongoClient); @Override Long getMaxMessageId(String topicName, String consumerId); @Override void add(String topicName, String consumerId, Long messageId, String sourceConsumerIp); }
AckDAOImpl implements AckDAO { @Override public Long getMaxMessageId(String topicName, String consumerId) { DBCollection collection = this.mongoClient.getAckCollection(topicName, consumerId); DBObject fields = BasicDBObjectBuilder.start().add(MSG_ID, Integer.valueOf(1)).get(); DBObject orderBy = BasicDBObjectBuilder.start().add(MSG_ID, Integer.valueOf(-1)).get(); DBCursor cursor = collection.find(new BasicDBObject(), fields).sort(orderBy).limit(1); try { if (cursor.hasNext()) { DBObject result = cursor.next(); BSONTimestamp timestamp = (BSONTimestamp) result.get(MSG_ID); return MongoUtils.BSONTimestampToLong(timestamp); } } finally { cursor.close(); } return null; }  void setMongoClient(MongoClient mongoClient); @Override Long getMaxMessageId(String topicName, String consumerId); @Override void add(String topicName, String consumerId, Long messageId, String sourceConsumerIp); static final String MSG_ID; static final String SRC_CONSUMER_IP; static final String TICK; }