components/kpi-computation-ms/src/main/java/org/onap/dcaegen2/kpi/dmaap/DmaapClient.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2021 China Mobile.
   4  *  Copyright (C) 2022 Wipro Limited.
   5  * ================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  *
  18  * SPDX-License-Identifier: Apache-2.0
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.dcaegen2.kpi.dmaap;
  23
  24 import com.att.nsa.cambria.client.CambriaConsumer;
  25
  26 import java.util.Map;
  27 import java.util.concurrent.Executors;
  28 import java.util.concurrent.ScheduledExecutorService;
  29 import java.util.concurrent.TimeUnit;
  30
  31 import javax.annotation.PostConstruct;
  32
  33 import org.onap.dcaegen2.kpi.models.Configuration;
  34 import org.onap.dcaegen2.kpi.utils.DmaapUtils;
  35 import org.slf4j.Logger;
  36 import org.slf4j.LoggerFactory;
  37
  38 import org.springframework.stereotype.Component;
  39
  40 /**
  41  * This class initializes and starts the dmaap client to listen on application
  42  * required dmaap events.
  43  */
  44 @Component
  45 public class DmaapClient {
  46
  47     private Configuration configuration;
  48     private static Logger log = LoggerFactory.getLogger(DmaapClient.class);
  49
  50     private DmaapUtils dmaapUtils;
  51
  52     /**
  53      * init dmaap client.
  54      */
  55     @PostConstruct
  56     public void initClient() {
  57         log.debug("initializing client");
  58         dmaapUtils = new DmaapUtils();
  59         configuration = Configuration.getInstance();
  60         if (log.isDebugEnabled()) {
  61             log.debug(configuration.toString());
  62         }
  63
  64         startClient();
  65     }
  66
  67     /**
  68      * start dmaap client.
  69      */
  70     @SuppressWarnings("unchecked")
  71     public synchronized void startClient() {
  72
  73         Map<String, Object> streamSubscribes = configuration.getStreamsSubscribes();
  74
  75         String pmTopicUrl = ((Map<String, String>) ((Map<String, Object>) streamSubscribes
  76                 .get("performance_management_topic")).get("dmaap_info")).get("topic_url");
  77         String[] pmTopicSplit = pmTopicUrl.split("\\/");
  78         String pmTopic = pmTopicSplit[pmTopicSplit.length - 1];
  79         log.debug("pm topic : {}", pmTopic);
  80
  81         CambriaConsumer pmNotifCambriaConsumer = dmaapUtils.buildConsumer(configuration, pmTopic);
  82
  83         ScheduledExecutorService executorPool;
  84
  85         // create notification consumers for PM
  86         NotificationConsumer pmNotificationConsumer = new NotificationConsumer(pmNotifCambriaConsumer,
  87                 new KpiComputationCallBack());
  88         // start pm notification consumer threads
  89         executorPool = Executors.newScheduledThreadPool(10);
  90         executorPool.scheduleAtFixedRate(pmNotificationConsumer, 0, configuration.getPollingInterval(),
  91                 TimeUnit.SECONDS);
  92
  93     }
  94
  95 }