@Test public void testSaveMessage() { SwallowMessage expectedMessage = createMessage(); expectedMessage.setContent("content in testSaveMessage"); messageDAO.saveMessage(TOPIC_NAME, expectedMessage); SwallowMessage actualMessage = messageDAO.getMaxMessage(TOPIC_NAME); Assert.assertTrue(expectedMessage.equalsWithoutMessageId(actualMessage)); }
@Override public void saveMessage(String topicName, SwallowMessage message) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); BasicDBObjectBuilder builder = BasicDBObjectBuilder.start().add(ID, new BSONTimestamp()); String content = message.getContent(); if (content != null && !"".equals(content.trim())) { builder.add(CONTENT, content); } Date generatedTime = message.getGeneratedTime(); if (generatedTime != null) { builder.add(GENERATED_TIME, generatedTime); } String version = message.getVersion(); if (version != null && !"".equals(version.trim())) { builder.add(VERSION, version); } Map<String, String> properties = message.getProperties(); if (properties != null && properties.size() > 0) { builder.add(PROPERTIES, properties); } Map<String, String> internalProperties = message.getInternalProperties(); if (internalProperties != null && internalProperties.size() > 0) { builder.add(INTERNAL_PROPERTIES, internalProperties); } String sha1 = message.getSha1(); if (sha1 != null && !"".equals(sha1.trim())) { builder.add(SHA1, sha1); } String type = message.getType(); if (type != null && !"".equals(type.trim())) { builder.add(TYPE, type); } String sourceIp = message.getSourceIp(); if (sourceIp != null && !"".equals(sourceIp.trim())) { builder.add(SOURCE_IP, sourceIp); } collection.insert(builder.get()); }
MessageDAOImpl implements MessageDAO { @Override public void saveMessage(String topicName, SwallowMessage message) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); BasicDBObjectBuilder builder = BasicDBObjectBuilder.start().add(ID, new BSONTimestamp()); String content = message.getContent(); if (content != null && !"".equals(content.trim())) { builder.add(CONTENT, content); } Date generatedTime = message.getGeneratedTime(); if (generatedTime != null) { builder.add(GENERATED_TIME, generatedTime); } String version = message.getVersion(); if (version != null && !"".equals(version.trim())) { builder.add(VERSION, version); } Map<String, String> properties = message.getProperties(); if (properties != null && properties.size() > 0) { builder.add(PROPERTIES, properties); } Map<String, String> internalProperties = message.getInternalProperties(); if (internalProperties != null && internalProperties.size() > 0) { builder.add(INTERNAL_PROPERTIES, internalProperties); } String sha1 = message.getSha1(); if (sha1 != null && !"".equals(sha1.trim())) { builder.add(SHA1, sha1); } String type = message.getType(); if (type != null && !"".equals(type.trim())) { builder.add(TYPE, type); } String sourceIp = message.getSourceIp(); if (sourceIp != null && !"".equals(sourceIp.trim())) { builder.add(SOURCE_IP, sourceIp); } collection.insert(builder.get()); } }
MessageDAOImpl implements MessageDAO { @Override public void saveMessage(String topicName, SwallowMessage message) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); BasicDBObjectBuilder builder = BasicDBObjectBuilder.start().add(ID, new BSONTimestamp()); String content = message.getContent(); if (content != null && !"".equals(content.trim())) { builder.add(CONTENT, content); } Date generatedTime = message.getGeneratedTime(); if (generatedTime != null) { builder.add(GENERATED_TIME, generatedTime); } String version = message.getVersion(); if (version != null && !"".equals(version.trim())) { builder.add(VERSION, version); } Map<String, String> properties = message.getProperties(); if (properties != null && properties.size() > 0) { builder.add(PROPERTIES, properties); } Map<String, String> internalProperties = message.getInternalProperties(); if (internalProperties != null && internalProperties.size() > 0) { builder.add(INTERNAL_PROPERTIES, internalProperties); } String sha1 = message.getSha1(); if (sha1 != null && !"".equals(sha1.trim())) { builder.add(SHA1, sha1); } String type = message.getType(); if (type != null && !"".equals(type.trim())) { builder.add(TYPE, type); } String sourceIp = message.getSourceIp(); if (sourceIp != null && !"".equals(sourceIp.trim())) { builder.add(SOURCE_IP, sourceIp); } collection.insert(builder.get()); }  }
MessageDAOImpl implements MessageDAO { @Override public void saveMessage(String topicName, SwallowMessage message) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); BasicDBObjectBuilder builder = BasicDBObjectBuilder.start().add(ID, new BSONTimestamp()); String content = message.getContent(); if (content != null && !"".equals(content.trim())) { builder.add(CONTENT, content); } Date generatedTime = message.getGeneratedTime(); if (generatedTime != null) { builder.add(GENERATED_TIME, generatedTime); } String version = message.getVersion(); if (version != null && !"".equals(version.trim())) { builder.add(VERSION, version); } Map<String, String> properties = message.getProperties(); if (properties != null && properties.size() > 0) { builder.add(PROPERTIES, properties); } Map<String, String> internalProperties = message.getInternalProperties(); if (internalProperties != null && internalProperties.size() > 0) { builder.add(INTERNAL_PROPERTIES, internalProperties); } String sha1 = message.getSha1(); if (sha1 != null && !"".equals(sha1.trim())) { builder.add(SHA1, sha1); } String type = message.getType(); if (type != null && !"".equals(type.trim())) { builder.add(TYPE, type); } String sourceIp = message.getSourceIp(); if (sourceIp != null && !"".equals(sourceIp.trim())) { builder.add(SOURCE_IP, sourceIp); } collection.insert(builder.get()); }  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 void saveMessage(String topicName, SwallowMessage message) { DBCollection collection = this.mongoClient.getMessageCollection(topicName); BasicDBObjectBuilder builder = BasicDBObjectBuilder.start().add(ID, new BSONTimestamp()); String content = message.getContent(); if (content != null && !"".equals(content.trim())) { builder.add(CONTENT, content); } Date generatedTime = message.getGeneratedTime(); if (generatedTime != null) { builder.add(GENERATED_TIME, generatedTime); } String version = message.getVersion(); if (version != null && !"".equals(version.trim())) { builder.add(VERSION, version); } Map<String, String> properties = message.getProperties(); if (properties != null && properties.size() > 0) { builder.add(PROPERTIES, properties); } Map<String, String> internalProperties = message.getInternalProperties(); if (internalProperties != null && internalProperties.size() > 0) { builder.add(INTERNAL_PROPERTIES, internalProperties); } String sha1 = message.getSha1(); if (sha1 != null && !"".equals(sha1.trim())) { builder.add(SHA1, sha1); } String type = message.getType(); if (type != null && !"".equals(type.trim())) { builder.add(TYPE, type); } String sourceIp = message.getSourceIp(); if (sourceIp != null && !"".equals(sourceIp.trim())) { builder.add(SOURCE_IP, sourceIp); } collection.insert(builder.get()); }  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; }