@Test public void testGetMessage() { SwallowMessage expectedMessage = createMessage(); expectedMessage.setContent("content in testGetMessage"); messageDAO.saveMessage(TOPIC_NAME, expectedMessage); Long maxMessageId = messageDAO.getMaxMessageId(TOPIC_NAME); SwallowMessage actualMessage = messageDAO.getMessage(TOPIC_NAME, maxMessageId); Assert.assertTrue(expectedMessage.equalsWithoutMessageId(actualMessage)); }
@Override public SwallowMessage getMessage(String topicName, Long messageId) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); DBObject query = BasicDBObjectBuilder.start().add(ID, MongoUtils.longToBSONTimestamp(messageId)).get(); DBObject result = collection.findOne(query); if (result != null) { SwallowMessage swallowMessage = new SwallowMessage(); try { convert(result, swallowMessage); return swallowMessage; } catch (RuntimeException e) { LOG.error("Error when convert resultset to SwallowMessage.", e); } } return null; }
MessageDAOImpl implements MessageDAO { @Override public SwallowMessage getMessage(String topicName, Long messageId) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); DBObject query = BasicDBObjectBuilder.start().add(ID, MongoUtils.longToBSONTimestamp(messageId)).get(); DBObject result = collection.findOne(query); if (result != null) { SwallowMessage swallowMessage = new SwallowMessage(); try { convert(result, swallowMessage); return swallowMessage; } catch (RuntimeException e) { LOG.error("Error when convert resultset to SwallowMessage.", e); } } return null; } }
MessageDAOImpl implements MessageDAO { @Override public SwallowMessage getMessage(String topicName, Long messageId) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); DBObject query = BasicDBObjectBuilder.start().add(ID, MongoUtils.longToBSONTimestamp(messageId)).get(); DBObject result = collection.findOne(query); if (result != null) { SwallowMessage swallowMessage = new SwallowMessage(); try { convert(result, swallowMessage); return swallowMessage; } catch (RuntimeException e) { LOG.error("Error when convert resultset to SwallowMessage.", e); } } return null; }  }
MessageDAOImpl implements MessageDAO { @Override public SwallowMessage getMessage(String topicName, Long messageId) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); DBObject query = BasicDBObjectBuilder.start().add(ID, MongoUtils.longToBSONTimestamp(messageId)).get(); DBObject result = collection.findOne(query); if (result != null) { SwallowMessage swallowMessage = new SwallowMessage(); try { convert(result, swallowMessage); return swallowMessage; } catch (RuntimeException e) { LOG.error("Error when convert resultset to SwallowMessage.", e); } } return null; }  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 SwallowMessage getMessage(String topicName, Long messageId) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); DBObject query = BasicDBObjectBuilder.start().add(ID, MongoUtils.longToBSONTimestamp(messageId)).get(); DBObject result = collection.findOne(query); if (result != null) { SwallowMessage swallowMessage = new SwallowMessage(); try { convert(result, swallowMessage); return swallowMessage; } catch (RuntimeException e) { LOG.error("Error when convert resultset to SwallowMessage.", e); } } return null; }  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; }