components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/DmaapService.java

   1 /*
   2 * ============LICENSE_START=======================================================
   3 * ONAP : DATALAKE
   4 * ================================================================================
   5 * Copyright 2019 China Mobile
   6 *=================================================================================
   7 * Licensed under the Apache License, Version 2.0 (the "License");
   8 * you may not use this file except in compliance with the License.
   9 * You may obtain a copy of the License at
  10 *
  11 *     http://www.apache.org/licenses/LICENSE-2.0
  12 *
  13 * Unless required by applicable law or agreed to in writing, software
  14 * distributed under the License is distributed on an "AS IS" BASIS,
  15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16 * See the License for the specific language governing permissions and
  17 * limitations under the License.
  18 * ============LICENSE_END=========================================================
  19 */
  20
  21 package org.onap.datalake.feeder.service;
  22
  23 import java.io.IOException;
  24 import java.util.ArrayList;
  25 import java.util.List;
  26
  27 import javax.annotation.PostConstruct;
  28
  29 import org.apache.zookeeper.WatchedEvent;
  30 import org.apache.zookeeper.Watcher;
  31 import org.apache.zookeeper.ZooKeeper;
  32 import org.onap.datalake.feeder.config.ApplicationConfiguration;
  33 import org.onap.datalake.feeder.domain.Topic;
  34 import org.slf4j.Logger;
  35 import org.slf4j.LoggerFactory;
  36 import org.springframework.beans.factory.annotation.Autowired;
  37 import org.springframework.stereotype.Service;
  38
  39 /**
  40  * This service will handle all the communication with Kafka
  41  *
  42  * @author Guobiao Mo
  43  *
  44  */
  45 @Service
  46 public class DmaapService {
  47
  48         private final Logger log = LoggerFactory.getLogger(this.getClass());
  49
  50         @Autowired
  51         private ApplicationConfiguration config;
  52
  53         @Autowired
  54         private TopicService topicService;
  55
  56
  57         @PostConstruct
  58         private void init() {
  59
  60         }
  61
  62         //get all topic names from Zookeeper
  63         public List<String> getTopics() {
  64                 try {
  65                         Watcher watcher = new Watcher() {
  66                                 @Override
  67                                 public void process(WatchedEvent event) {
  68                                         // TODO monitor new topics
  69
  70                                 }
  71                 };
  72                         ZooKeeper zk = new ZooKeeper(config.getDmaapZookeeperHostPort(), 10000, watcher);
  73                 List<String> topics = zk.getChildren("/brokers/topics", false);
  74                 return topics;
  75                 } catch (Exception e) {
  76                         log.error("Can not get topic list from Zookeeper, for testing, going to use hard coded topic list.", e);
  77                         return null;
  78                 }
  79         }
  80
  81         public List<String> getActiveTopics() throws IOException {
  82                 List<String> allTopics = new ArrayList<>(getTopics());
  83
  84                 List<String> ret = new ArrayList<>();
  85                 for (String topicStr : allTopics) {
  86                         Topic topic = topicService.getEffectiveTopic(topicStr, true);
  87                         if (topic.isEnabled()) {
  88                                 ret.add(topicStr);
  89                         }
  90                 }
  91                 return ret;
  92         }
  93
  94 }