policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumerImpl.java

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * O-RAN-SC
   4  * %%
   5  * Copyright (C) 2019 Nordix Foundation
   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.oransc.policyagent.dmaap;
  22
  23 import java.io.IOException;
  24 import java.util.Properties;
  25
  26 import org.onap.dmaap.mr.client.MRClientFactory;
  27 import org.onap.dmaap.mr.client.MRConsumer;
  28 import org.onap.dmaap.mr.client.response.MRConsumerResponse;
  29 import org.oransc.policyagent.configuration.ApplicationConfig;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32 import org.springframework.beans.factory.annotation.Autowired;
  33 import org.springframework.scheduling.annotation.EnableScheduling;
  34 import org.springframework.scheduling.annotation.Scheduled;
  35 import org.springframework.stereotype.Component;
  36
  37 @Component
  38 @EnableScheduling
  39 public class DmaapMessageConsumerImpl implements DmaapMessageConsumer {
  40
  41     private static final Logger logger = LoggerFactory.getLogger(DmaapMessageConsumerImpl.class);
  42
  43     private boolean alive = false;
  44     private final ApplicationConfig applicationConfig;
  45     protected MRConsumer consumer;
  46     private MRConsumerResponse response = null;
  47
  48     @Autowired
  49     public DmaapMessageConsumerImpl(ApplicationConfig applicationConfig) {
  50         this.applicationConfig = applicationConfig;
  51     }
  52
  53     @Scheduled(fixedRate = 1000 * 60)
  54     @Override
  55     public void run() {
  56         if (!alive) {
  57             init();
  58         }
  59         if (this.alive) {
  60             try {
  61                 Iterable<String> dmaapMsgs = fetchAllMessages();
  62                 for (String msg : dmaapMsgs) {
  63                     processMsg(msg);
  64                 }
  65             } catch (Exception e) {
  66                 logger.error("{}: cannot fetch because of ", this, e.getMessage(), e);
  67             }
  68         }
  69     }
  70
  71     private Iterable<String> fetchAllMessages() {
  72         response = consumer.fetchWithReturnConsumerResponse();
  73         if (response == null) {
  74             logger.warn("{}: DMaaP NULL response received", this);
  75         } else {
  76             logger.debug("DMaaP consumer received {} : {}", response.getResponseCode(), response.getResponseMessage());
  77             if (!"200".equals(response.getResponseCode())) {
  78                 logger.error("DMaaP consumer received: {} : {}", response.getResponseCode(),
  79                     response.getResponseMessage());
  80             }
  81         }
  82         return response.getActualMessages();
  83     }
  84
  85     @Override
  86     public void init() {
  87         Properties dmaapConsumerProperties = applicationConfig.getDmaapConsumerConfig();
  88         Properties dmaapPublisherProperties = applicationConfig.getDmaapPublisherConfig();
  89         // No need to start if there is no configuration.
  90         if (dmaapConsumerProperties == null || dmaapPublisherProperties == null || dmaapConsumerProperties.size() == 0
  91             || dmaapPublisherProperties.size() == 0) {
  92             return;
  93         }
  94         // Do we need to do any validation of properties before calling the factory?
  95         try {
  96             consumer = MRClientFactory.createConsumer(dmaapConsumerProperties);
  97             this.alive = true;
  98         } catch (IOException e) {
  99             logger.error("Exception occurred while creating Dmaap Consumer", e);
 100         }
 101     }
 102
 103     @Override
 104     public void processMsg(String msg) throws Exception {
 105         System.out.println("sysout" + msg);
 106         // Call the concurrent Task executor to handle the incoming request
 107         // Validate the Input & if its valid, post the ACCEPTED Response back to DMAAP
 108         // through REST CLIENT
 109         // Call the Controller with the extracted payload
 110     }
 111
 112     @Override
 113     public boolean isAlive() {
 114         return alive;
 115     }
 116
 117     @Override
 118     public void stopConsumer() {
 119         alive = false;
 120     }
 121 }