`save_raw` bit(1) DEFAULT NULL,\r
`ttl` int(11) DEFAULT NULL,\r
`data_format` varchar(255) DEFAULT NULL,\r
- `default_topic` varchar(255) DEFAULT NULL,\r
- PRIMARY KEY (`name`),\r
- KEY `FK_default_topic` (`default_topic`),\r
- CONSTRAINT `FK_default_topic` FOREIGN KEY (`default_topic`) REFERENCES `topic` (`name`)\r
+ PRIMARY KEY (`name`)\r
) ENGINE=InnoDB DEFAULT CHARSET=utf8;\r
\r
\r
-- in production, default enabled should be off\r
insert into `topic`(`name`,`enabled`,`save_raw`,`ttl`,`data_format`) values ('_DL_DEFAULT_',1,0,3650,'JSON');\r
insert into `topic`(`name`,`enabled`) values ('__consumer_offsets',0);\r
-insert into `topic`(`name`,correlate_cleared_message,`enabled`,default_topic, message_id_path) values ('unauthenticated.SEC_FAULT_OUTPUT',1,0,'_DL_DEFAULT_', '/event/commonEventHeader/eventName,/event/commonEventHeader/reportingEntityName,/event/faultFields/specificProblem');\r
+insert into `topic`(`name`,correlate_cleared_message,`enabled`, message_id_path) values ('unauthenticated.SEC_FAULT_OUTPUT',1,0,'/event/commonEventHeader/eventName,/event/commonEventHeader/reportingEntityName,/event/faultFields/specificProblem');\r
\r
\r
insert into `map_db_topic`(`db_name`,`topic_name`) values ('Couchbase','_DL_DEFAULT_');\r
private String dmaapKafkaHostPort;
private String dmaapKafkaGroup;
private long dmaapKafkaTimeout;
+ private String[] dmaapKafkaExclude;
private int dmaapCheckNewTopicIntervalInSec;
/**
* This controller manages topic settings.
*
- * Topic "_DL_DEFAULT_" acts as the default. For example, if a topic's
- * enabled=null, _DL_DEFAULT_.enabled is used for that topic. All the settings
- * are saved in database. topic "_DL_DEFAULT_" is populated at setup by a DB
- * script.
+ * Topic "_DL_DEFAULT_" acts as the default.
+ * If a topic is not present in database, "_DL_DEFAULT_" is used for it.
+ * If a topic is present in database, itself should be complete, and no setting from "_DL_DEFAULT_" is used.
+ * Topic "_DL_DEFAULT_" is populated at setup by a DB script.
*
* @author Guobiao Mo
* @contributor Kate Hsuan @ QCT
@GetMapping("")
@ResponseBody
- @ApiOperation(value="List all topics in database")
+ @ApiOperation(value="List all topic names in database")
public List<String> list() {
Iterable<Topic> ret = topicRepository.findAll();
List<String> retString = new ArrayList<>();
sendError(response, 400, "Topic already exists "+topicConfig.getName());
return null;
} else {
- PostReturnBody<TopicConfig> retBody = new PostReturnBody<>();
Topic wTopic = topicService.fillTopicConfiguration(topicConfig);
if(wTopic.getTtl() == 0)
wTopic.setTtl(3650);
- topicRepository.save(wTopic);
- mkPostReturnBody(retBody, 200, wTopic);
- return retBody;
+ topicRepository.save(wTopic);
+ return mkPostReturnBody(200, wTopic);
}
}
sendError(response, 404, "Topic not found "+topicConfig.getName());
return null;
} else {
- PostReturnBody<TopicConfig> retBody = new PostReturnBody<>();
topicService.fillTopicConfiguration(topicConfig, oldTopic);
topicRepository.save(oldTopic);
- mkPostReturnBody(retBody, 200, oldTopic);
- return retBody;
+ return mkPostReturnBody(200, oldTopic);
}
}
- private void mkPostReturnBody(PostReturnBody<TopicConfig> retBody, int statusCode, Topic topic)
+ private PostReturnBody<TopicConfig> mkPostReturnBody(int statusCode, Topic topic)
{
+ PostReturnBody<TopicConfig> retBody = new PostReturnBody<>();
retBody.setStatusCode(statusCode);
retBody.setReturnBody(topic.getTopicConfig());
+
+ return retBody;
}
private void sendError(HttpServletResponse response, int sc, String msg) throws IOException {
import javax.persistence.ManyToMany;
import javax.persistence.Table;
-import org.apache.commons.lang3.StringUtils;
-import org.json.JSONObject;
import org.onap.datalake.feeder.dto.TopicConfig;
-import org.onap.datalake.feeder.enumeration.DataFormat;
import com.fasterxml.jackson.annotation.JsonBackReference;
@Table(name = "topic")
public class Topic {
@Id
- @Column(name="`name`")
+ @Column(name = "`name`")
private String name;//topic name
-
- //for protected Kafka topics
+ //for protected Kafka topics
@Column(name = "`login`")
private String login;
@JsonBackReference
//@JsonManagedReference
@ManyToMany(fetch = FetchType.EAGER)
- @JoinTable( name = "map_db_topic",
- joinColumns = { @JoinColumn(name="topic_name") },
- inverseJoinColumns = { @JoinColumn(name="db_name") }
- )
+ @JoinTable(name = "map_db_topic", joinColumns = { @JoinColumn(name = "topic_name") }, inverseJoinColumns = { @JoinColumn(name = "db_name") })
protected Set<Db> dbs;
/**
* indicate if we should monitor this topic
*/
- @Column(name="`enabled`")
+ @Column(name = "`enabled`")
private Boolean enabled;
/**
private Boolean saveRaw;
/**
- * need to explicitly tell feeder the data format of the message.
- * support JSON, XML, YAML, TEXT
+ * need to explicitly tell feeder the data format of the message. support JSON,
+ * XML, YAML, TEXT
*/
- @Column(name="`data_format`")
+ @Column(name = "`data_format`")
private String dataFormat;
/**
this.name = name;
}
- public Topic clone() { //TODO will use TopicConfig
- Topic ret = new Topic();
- ret.setCorrelateClearedMessage(correlateClearedMessage);
- ret.setDataFormat(dataFormat);
- ret.setDbs(dbs);
- ret.setEnabled(enabled);
- ret.setLogin(login);
- ret.setMessageIdPath(messageIdPath);
- ret.setName(name);
- ret.setPass(pass);
- ret.setSaveRaw(saveRaw);
- ret.setTtl(ttl);
-
- return ret;
- }
-
public boolean isDefault() {
return "_DL_DEFAULT_".equals(name);
}
public int getTtl() {
if (ttl != null) {
return ttl;
- } else {
+ } else {
return 3650;//default to 10 years for safe
}
}
- public DataFormat getDataFormat() {
- if (dataFormat != null) {
- return DataFormat.fromString(dataFormat);
- } else {
- return null;
- }
- }
-
private boolean is(Boolean b) {
return is(b, false);
}
private boolean is(Boolean b, boolean defaultValue) {
if (b != null) {
return b;
- } else {
+ } else {
return defaultValue;
}
}
return is(saveRaw);
}
- public boolean supportElasticsearch() {
- return containDb("Elasticsearch");//TODO string hard codes
- }
-
- public boolean supportCouchbase() {
- return containDb("Couchbase");
- }
-
- public boolean supportDruid() {
- return containDb("Druid");
- }
-
- public boolean supportMongoDB() {
- return containDb("MongoDB");
- }
-
- private boolean containDb(String dbName) {
- Db db = new Db(dbName);
-
- if (dbs != null && dbs.contains(db)) {
- return true;
- } else {
- return false;
- }
- }
-
- //extract DB id from JSON attributes, support multiple attributes
- public String getMessageId(JSONObject json) {
- String id = null;
-
- if (StringUtils.isNotBlank(messageIdPath)) {
- String[] paths = messageIdPath.split(",");
-
- StringBuilder sb = new StringBuilder();
- for (int i = 0; i < paths.length; i++) {
- if (i > 0) {
- sb.append('^');
- }
- sb.append(json.query(paths[i]).toString());
- }
- id = sb.toString();
- }
-
- return id;
- }
-
public TopicConfig getTopicConfig() {
TopicConfig tConfig = new TopicConfig();
-
+
tConfig.setName(getName());
- tConfig.setEnable(getEnabled());
- if(getDataFormat() != null)
- tConfig.setDataFormat(getDataFormat().toString());
- tConfig.setSaveRaw(getSaveRaw());
- tConfig.setCorrelatedClearredMessage((getCorrelateClearedMessage() == null) ? getCorrelateClearedMessage() : false);
+ tConfig.setEnabled(isEnabled());
+ tConfig.setDataFormat(dataFormat);
+ tConfig.setSaveRaw(isSaveRaw());
+ tConfig.setCorrelateClearedMessage(isCorrelateClearedMessage());
tConfig.setMessageIdPath(getMessageIdPath());
tConfig.setTtl(getTtl());
Set<Db> topicDb = getDbs();
List<String> dbList = new ArrayList<>();
- for(Db item: topicDb)
- {
- dbList.add(item.getName());
+ if (topicDb != null) {
+ for (Db item : topicDb) {
+ dbList.add(item.getName());
+ }
}
tConfig.setSinkdbs(dbList);
-
+
return tConfig;
}
import lombok.Getter;
import lombok.Setter;
-import org.onap.datalake.feeder.domain.Topic;
-import org.onap.datalake.feeder.domain.Db;
-import org.onap.datalake.feeder.repository.DbRepository;
-import java.util.HashSet;
import java.util.List;
-import java.util.Set;
+import org.apache.commons.lang3.StringUtils;
+import org.json.JSONObject;
+import org.onap.datalake.feeder.enumeration.DataFormat;
/**
* JSON request body for Topic manipulation.
public class TopicConfig {
- private String name;
- private String login;
- private String password;
- private List<String> sinkdbs;
- private boolean enable;
- private boolean saveRaw;
- private String dataFormat;
- private int ttl;
- private boolean correlatedClearredMessage;
- private String messageIdPath;
+ private String name;
+ private String login;
+ private String password;
+ private List<String> sinkdbs;
+ private boolean enabled;
+ private boolean saveRaw;
+ private String dataFormat;
+ private int ttl;
+ private boolean correlateClearedMessage;
+ private String messageIdPath;
+ public DataFormat getDataFormat2() {
+ if (dataFormat != null) {
+ return DataFormat.fromString(dataFormat);
+ } else {
+ return null;
+ }
+ }
+
+ public boolean supportElasticsearch() {
+ return containDb("Elasticsearch");//TODO string hard codes
+ }
+
+ public boolean supportCouchbase() {
+ return containDb("Couchbase");
+ }
+
+ public boolean supportDruid() {
+ return containDb("Druid");
+ }
+
+ public boolean supportMongoDB() {
+ return containDb("MongoDB");
+ }
+
+ private boolean containDb(String dbName) {
+ return (sinkdbs != null && sinkdbs.contains(dbName));
+ }
+
+ //extract DB id from JSON attributes, support multiple attributes
+ public String getMessageId(JSONObject json) {
+ String id = null;
+
+ if (StringUtils.isNotBlank(messageIdPath)) {
+ String[] paths = messageIdPath.split(",");
+
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < paths.length; i++) {
+ if (i > 0) {
+ sb.append('^');
+ }
+ sb.append(json.query(paths[i]).toString());
+ }
+ id = sb.toString();
+ }
+
+ return id;
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == null)
+ return false;
+
+ if (this.getClass() != obj.getClass())
+ return false;
+
+ return name.equals(((TopicConfig) obj).getName());
+ }
+
+ @Override
+ public int hashCode() {
+ return name.hashCode();
+ }
}
import org.onap.datalake.feeder.config.ApplicationConfiguration;
import org.onap.datalake.feeder.domain.Db;
import org.onap.datalake.feeder.domain.Topic;
+import org.onap.datalake.feeder.dto.TopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
bucket.close();
}
- public void saveJsons(Topic topic, List<JSONObject> jsons) {
+ public void saveJsons(TopicConfig topic, List<JSONObject> jsons) {
List<JsonDocument> documents= new ArrayList<>(jsons.size());
for(JSONObject json : jsons) {
//convert to Couchbase JsonObject from org.json JSONObject
log.debug("saved text to topic = {}, this batch count = {} ", topic, documents.size());
}
- public String getId(Topic topic, JSONObject json) {
+ public String getId(TopicConfig topic, JSONObject json) {
//if this topic requires extract id from JSON
String id = topic.getMessageId(json);
if(id != null) {
import java.util.Collections;
import java.util.List;
-import javax.annotation.PostConstruct;
-
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
-import org.onap.datalake.feeder.domain.Topic;
+import org.onap.datalake.feeder.dto.TopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@Autowired
private TopicService topicService;
-
- @PostConstruct
- private void init() {
-
- }
-
//get all topic names from Zookeeper
public List<String> getTopics() {
try {
@Override
public void process(WatchedEvent event) {
// TODO monitor new topics
-
+
}
- };
+ };
ZooKeeper zk = new ZooKeeper(config.getDmaapZookeeperHostPort(), 10000, watcher);
- List<String> topics = zk.getChildren("/brokers/topics", false);
- return topics;
+ List<String> topics = zk.getChildren("/brokers/topics", false);
+ String[] excludes = config.getDmaapKafkaExclude();
+ for (String exclude : excludes) {
+ topics.remove(exclude);
+ }
+ return topics;
} catch (Exception e) {
log.error("Can not get topic list from Zookeeper, for testing, going to use hard coded topic list.", e);
- return null;
+ return Collections.emptyList();
}
- }
+ }
- public List<String> getActiveTopics() throws IOException {
+ public List<String> getActiveTopics() throws IOException {
List<String> allTopics = getTopics();
- if(allTopics == null) {
+ if (allTopics == null) {
return Collections.emptyList();
}
List<String> ret = new ArrayList<>();
for (String topicStr : allTopics) {
- Topic topic = topicService.getEffectiveTopic(topicStr, true);
- if (topic.isEnabled()) {
+ TopicConfig topicConfig = topicService.getEffectiveTopic(topicStr, true);
+ if (topicConfig.isEnabled()) {
ret.add(topicStr);
}
}
import org.json.JSONObject;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
import org.onap.datalake.feeder.domain.Db;
-import org.onap.datalake.feeder.domain.Topic;
+import org.onap.datalake.feeder.dto.TopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);
if(!exists){
+ //TODO submit mapping template
CreateIndexRequest createIndexRequest = new CreateIndexRequest(topicLower);
CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
log.info("{} : created {}", createIndexResponse.index(), createIndexResponse.isAcknowledged());
}
//TTL is not supported in Elasticsearch 5.0 and later, what can we do? FIXME
- public void saveJsons(Topic topic, List<JSONObject> jsons) {
+ public void saveJsons(TopicConfig topic, List<JSONObject> jsons) {
BulkRequest request = new BulkRequest();
for (JSONObject json : jsons) {
request.add(new IndexRequest(topic.getName().toLowerCase(), config.getElasticsearchType(), id).source(json.toString(), XContentType.JSON));
}
+
+ log.debug("saving text to topic = {}, batch count = {} ", topic, jsons.size());
+
if(config.isAsync()) {
client.bulkAsync(request, RequestOptions.DEFAULT, listener);
}else {
* The search API can only query all data or based on the fields in the source.
* So use the get API, three parameters: index, type, document id
*/
- private boolean correlateClearedMessage(Topic topic, JSONObject json) {
+ private boolean correlateClearedMessage(TopicConfig topic, JSONObject json) {
boolean found = false;
String eName = null;
import org.json.JSONObject;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
import org.onap.datalake.feeder.domain.Db;
-import org.onap.datalake.feeder.domain.Topic;
+import org.onap.datalake.feeder.dto.TopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.mongodb.ServerAddress;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
-import com.mongodb.DB;
/**
* Service for using MongoDB
mongoClient.close();
}
- public void saveJsons(Topic topic, List<JSONObject> jsons) {
+ public void saveJsons(TopicConfig topic, List<JSONObject> jsons) {
if(dbReady == false)
return;
List<Document> documents = new ArrayList<>(jsons.size());
MongoCollection<Document> collection = mongoCollectionMap.computeIfAbsent(collectionName, k -> database.getCollection(k));
collection.insertMany(documents);
- log.debug("saved text to topic = {}, topic total count = {} ", topic, collection.countDocuments());
+ log.debug("saved text to topic = {}, batch count = {} ", topic, jsons.size());
}
}
private void init() {
async = config.isAsync();
Properties consumerConfig = getConsumerConfig();
+ log.info("Kafka ConsumerConfig: {}", consumerConfig);
consumer = new KafkaConsumer<>(consumerConfig);
}
import org.json.XML;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
import org.onap.datalake.feeder.domain.Topic;
+import org.onap.datalake.feeder.dto.TopicConfig;
import org.onap.datalake.feeder.enumeration.DataFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@Autowired
private ElasticsearchService elasticsearchService;
- private Map<String, Topic> topicMap = new HashMap<>();
+ private Map<String, TopicConfig> topicMap = new HashMap<>();
private ObjectMapper yamlReader;
return;
}
- Topic topic = topicMap.computeIfAbsent(topicStr, k -> { //TODO get topic updated settings from DB periodically
+ TopicConfig topic = topicMap.computeIfAbsent(topicStr, k -> { //TODO get topic updated settings from DB periodically
return topicService.getEffectiveTopic(topicStr);
});
saveJsons(topic, docs);
}
- private JSONObject messageToJson(Topic topic, Pair<Long, String> pair) throws JSONException, JsonParseException, JsonMappingException, IOException {
+ private JSONObject messageToJson(TopicConfig topic, Pair<Long, String> pair) throws JSONException, JsonParseException, JsonMappingException, IOException {
long timestamp = pair.getLeft();
String text = pair.getRight();
JSONObject json = null;
- DataFormat dataFormat = topic.getDataFormat();
+ DataFormat dataFormat = topic.getDataFormat2();
switch (dataFormat) {
case JSON:
return json;
}
- private void saveJsons(Topic topic, List<JSONObject> jsons) {
+ private void saveJsons(TopicConfig topic, List<JSONObject> jsons) {
if (topic.supportMongoDB()) {
mongodbService.saveJsons(topic, jsons);
}
@Autowired
private DbRepository dbRepository;
- public Topic getEffectiveTopic(String topicStr) {
+ public TopicConfig getEffectiveTopic(String topicStr) {
try {
return getEffectiveTopic(topicStr, false);
} catch (IOException e) {
return null;
}
- public Topic getEffectiveTopic(String topicStr, boolean ensureTableExist) throws IOException {
+ public TopicConfig getEffectiveTopic(String topicStr, boolean ensureTableExist) throws IOException {
Topic topic = getTopic(topicStr);
if (topic == null) {
- topic = getDefaultTopic().clone();
- topic.setName(topicStr);
+ topic = getDefaultTopic();
}
+ TopicConfig topicConfig = topic.getTopicConfig();
+ topicConfig.setName(topicStr);//need to change name if it comes from DefaultTopic
- if(ensureTableExist && topic.isEnabled() && topic.supportElasticsearch()) {
+ if(ensureTableExist && topicConfig.isEnabled() && topicConfig.supportElasticsearch()) {
elasticsearchService.ensureTableExist(topicStr);
}
- return topic;
+ return topicConfig;
}
public Topic getTopic(String topicStr) {
topic.setName(tConfig.getName());
topic.setLogin(tConfig.getLogin());
topic.setPass(tConfig.getPassword());
- topic.setEnabled(tConfig.isEnable());
+ topic.setEnabled(tConfig.isEnabled());
topic.setSaveRaw(tConfig.isSaveRaw());
topic.setTtl(tConfig.getTtl());
- topic.setCorrelateClearedMessage(tConfig.isCorrelatedClearredMessage());
+ topic.setCorrelateClearedMessage(tConfig.isCorrelateClearedMessage());
topic.setDataFormat(tConfig.getDataFormat());
topic.setMessageIdPath(tConfig.getMessageIdPath());
dmaapKafkaHostPort=message-router-kafka:9092
dmaapKafkaGroup=dlgroup10
dmaapKafkaTimeout=60
+dmaapKafkaExclude[0]=__consumer_offsets
+dmaapKafkaExclude[1]=msgrtr.apinode.metrics.dmaap
#check for new topics
dmaapCheckNewTopicIntervalInSec=3000
when(topicRepository.findById("a")).thenReturn(Optional.of(a));
TopicConfig ac = new TopicConfig();
ac.setName("a");
- ac.setEnable(true);
+ ac.setEnabled(true);
PostReturnBody<TopicConfig> postConfig1 = topicController.updateTopic("a", ac, mockBindingResult, httpServletResponse);
assertEquals(200, postConfig1.getStatusCode());
TopicConfig ret = postConfig1.getReturnBody();
assertEquals("a", ret.getName());
- assertEquals(true, ret.isEnable());
+ assertEquals(true, ret.isEnabled());
when(mockBindingResult.hasErrors()).thenReturn(true);
PostReturnBody<TopicConfig> postConfig2 = topicController.updateTopic("a", ac, mockBindingResult, httpServletResponse);
assertEquals(null, postConfig2);
*/
package org.onap.datalake.feeder.domain;
-import org.json.JSONObject;
import org.junit.Test;
import org.onap.datalake.feeder.enumeration.DataFormat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
/**
public class TopicTest {
- @Test
- public void getMessageId() {
- String text = "{ data: { data2 : { value : 'hello'}}}";
-
- JSONObject json = new JSONObject(text);
-
- Topic topic = new Topic("test getMessageId");
- topic.setMessageIdPath("/data/data2/value");
-
- String value = topic.getMessageId(json);
-
- assertEquals(value, "hello");
- }
-
@Test
public void getMessageIdFromMultipleAttributes() {
- String text = "{ data: { data2 : { value : 'hello'}, data3 : 'world'}}";
-
- JSONObject json = new JSONObject(text);
-
- Topic topic = new Topic("test getMessageId");
- topic.setMessageIdPath("/data/data2/value,/data/data3");
-
- String value = topic.getMessageId(json);
- assertEquals(value, "hello^world");
-
- topic.setMessageIdPath("");
- assertNull(topic.getMessageId(json));
-
+ Topic topic = new Topic("test getMessageId");
Topic defaultTopic = new Topic("_DL_DEFAULT_");
Topic testTopic = new Topic("test");
defaultTopic.setDbs(new HashSet<>());
defaultTopic.getDbs().add(new Db("Elasticsearch"));
- assertTrue(defaultTopic.supportElasticsearch());
- assertFalse(testTopic.supportCouchbase());
- assertFalse(testTopic.supportDruid());
- assertFalse(testTopic.supportMongoDB());
-
- defaultTopic.getDbs().remove(new Db("Elasticsearch"));
- assertFalse(testTopic.supportElasticsearch());
assertEquals(defaultTopic.getDataFormat(), null);
defaultTopic.setCorrelateClearedMessage(true);
assertTrue(defaultTopic.isEnabled());
assertTrue(defaultTopic.isSaveRaw());
- assertEquals(defaultTopic.getDataFormat(), DataFormat.XML);
+ assertEquals(defaultTopic.getTopicConfig().getDataFormat2(), DataFormat.XML);
defaultTopic.setDataFormat(null);
assertEquals(testTopic.getDataFormat(), null);
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP : DataLake
+ * ================================================================================
+ * Copyright 2019 China Mobile
+ *=================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.datalake.feeder.dto;
+
+import org.json.JSONObject;
+import org.junit.Test;
+import org.onap.datalake.feeder.domain.Db;
+import org.onap.datalake.feeder.domain.Topic;
+
+import java.util.HashSet;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Test Topic
+ *
+ * @author Guobiao Mo
+ */
+
+public class TopicConfigTest {
+
+ @Test
+ public void getMessageId() {
+ String text = "{ data: { data2 : { value : 'hello'}}}";
+
+ JSONObject json = new JSONObject(text);
+
+ Topic topic = new Topic("test getMessageId");
+ topic.setMessageIdPath("/data/data2/value");
+
+ TopicConfig topicConfig = topic.getTopicConfig();
+
+ String value = topicConfig.getMessageId(json);
+
+ assertEquals(value, "hello");
+ }
+
+ @Test
+ public void getMessageIdFromMultipleAttributes() {
+ String text = "{ data: { data2 : { value : 'hello'}, data3 : 'world'}}";
+
+ JSONObject json = new JSONObject(text);
+
+ Topic topic = new Topic("test getMessageId");
+ topic.setMessageIdPath("/data/data2/value,/data/data3");
+
+ TopicConfig topicConfig = topic.getTopicConfig();
+
+ String value = topicConfig.getMessageId(json);
+ assertEquals(value, "hello^world");
+
+ topic.setMessageIdPath("");
+ topicConfig = topic.getTopicConfig();
+ assertNull(topicConfig.getMessageId(json));
+
+ }
+
+ @Test
+ public void testIs() {
+ Topic testTopic = new Topic("test");
+
+ assertTrue(testTopic.equals(new Topic("test")));
+ assertEquals(testTopic.hashCode(), (new Topic("test")).hashCode());
+
+ testTopic.setDbs(new HashSet<>());
+ testTopic.getDbs().add(new Db("Elasticsearch"));
+
+ TopicConfig testTopicConfig = testTopic.getTopicConfig();
+
+ assertTrue(testTopicConfig.supportElasticsearch());
+ assertFalse(testTopicConfig.supportCouchbase());
+ assertFalse(testTopicConfig.supportDruid());
+ assertFalse(testTopicConfig.supportMongoDB());
+
+ testTopic.getDbs().remove(new Db("Elasticsearch"));
+ testTopicConfig = testTopic.getTopicConfig();
+ assertFalse(testTopicConfig.supportElasticsearch());
+
+ }
+}
CouchbaseService couchbaseService = new CouchbaseService();
couchbaseService.bucket = bucket;
couchbaseService.config = appConfig;
- couchbaseService.saveJsons(topic, jsons);
+ couchbaseService.saveJsons(topic.getTopicConfig(), jsons);
}
CouchbaseService couchbaseService = new CouchbaseService();
couchbaseService.bucket = bucket;
couchbaseService.config = appConfig;
- couchbaseService.saveJsons(topic, jsons);
+ couchbaseService.saveJsons(topic.getTopicConfig(), jsons);
}
@Test
when(config.getElasticsearchType()).thenReturn("doc");
when(config.isAsync()).thenReturn(true);
- elasticsearchService.saveJsons(topic, jsons);
+ elasticsearchService.saveJsons(topic.getTopicConfig(), jsons);
}
}
\ No newline at end of file
jsons.add(jsonObject);
jsons.add(jsonObject2);
- mongodbService.saveJsons(topic, jsons);
+ mongodbService.saveJsons(topic.getTopicConfig(), jsons);
}
}
\ No newline at end of file
Code Review / dcaegen2 / services.git / commitdiff