@Test public void testAdd() { 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 void add(String topicName, String consumerId, Long messageId, String sourceConsumerIp) { DBCollection collection = this.mongoClient.getAckCollection(topicName, consumerId); BSONTimestamp timestamp = MongoUtils.longToBSONTimestamp(messageId); Date curTime = new Date(); try { DBObject add = BasicDBObjectBuilder.start().add(MSG_ID, timestamp).add(SRC_CONSUMER_IP, sourceConsumerIp) .add(TICK, curTime).get(); collection.insert(add); } catch (MongoException e) { if (e.getMessage() != null && e.getMessage().indexOf("duplicate key") >= 0 || e.getCode() == 11000) { LOG.warn(e.getMessage() + ": _id is " + timestamp); } else { throw e; } } }
AckDAOImpl implements AckDAO { @Override public void add(String topicName, String consumerId, Long messageId, String sourceConsumerIp) { DBCollection collection = this.mongoClient.getAckCollection(topicName, consumerId); BSONTimestamp timestamp = MongoUtils.longToBSONTimestamp(messageId); Date curTime = new Date(); try { DBObject add = BasicDBObjectBuilder.start().add(MSG_ID, timestamp).add(SRC_CONSUMER_IP, sourceConsumerIp) .add(TICK, curTime).get(); collection.insert(add); } catch (MongoException e) { if (e.getMessage() != null && e.getMessage().indexOf("duplicate key") >= 0 || e.getCode() == 11000) { LOG.warn(e.getMessage() + ": _id is " + timestamp); } else { throw e; } } } }
AckDAOImpl implements AckDAO { @Override public void add(String topicName, String consumerId, Long messageId, String sourceConsumerIp) { DBCollection collection = this.mongoClient.getAckCollection(topicName, consumerId); BSONTimestamp timestamp = MongoUtils.longToBSONTimestamp(messageId); Date curTime = new Date(); try { DBObject add = BasicDBObjectBuilder.start().add(MSG_ID, timestamp).add(SRC_CONSUMER_IP, sourceConsumerIp) .add(TICK, curTime).get(); collection.insert(add); } catch (MongoException e) { if (e.getMessage() != null && e.getMessage().indexOf("duplicate key") >= 0 || e.getCode() == 11000) { LOG.warn(e.getMessage() + ": _id is " + timestamp); } else { throw e; } } }  }
AckDAOImpl implements AckDAO { @Override public void add(String topicName, String consumerId, Long messageId, String sourceConsumerIp) { DBCollection collection = this.mongoClient.getAckCollection(topicName, consumerId); BSONTimestamp timestamp = MongoUtils.longToBSONTimestamp(messageId); Date curTime = new Date(); try { DBObject add = BasicDBObjectBuilder.start().add(MSG_ID, timestamp).add(SRC_CONSUMER_IP, sourceConsumerIp) .add(TICK, curTime).get(); collection.insert(add); } catch (MongoException e) { if (e.getMessage() != null && e.getMessage().indexOf("duplicate key") >= 0 || e.getCode() == 11000) { LOG.warn(e.getMessage() + ": _id is " + timestamp); } else { throw e; } } }  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 void add(String topicName, String consumerId, Long messageId, String sourceConsumerIp) { DBCollection collection = this.mongoClient.getAckCollection(topicName, consumerId); BSONTimestamp timestamp = MongoUtils.longToBSONTimestamp(messageId); Date curTime = new Date(); try { DBObject add = BasicDBObjectBuilder.start().add(MSG_ID, timestamp).add(SRC_CONSUMER_IP, sourceConsumerIp) .add(TICK, curTime).get(); collection.insert(add); } catch (MongoException e) { if (e.getMessage() != null && e.getMessage().indexOf("duplicate key") >= 0 || e.getCode() == 11000) { LOG.warn(e.getMessage() + ": _id is " + timestamp); } else { throw e; } } }  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; }