src/main/java/org/onap/nbi/apis/hub/service/dmaap/CheckDMaaPEventsManager.java

   1 /**
   2  * Copyright (c) 2019 Huawei
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   5  * in compliance with the License. You may obtain a copy of the License at
   6  *
   7  * http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software distributed under the License
  10  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  11  * or implied. See the License for the specific language governing permissions and limitations under
  12  * the License.
  13  */
  14
  15 package org.onap.nbi.apis.hub.service.dmaap;
  16
  17 import java.io.IOException;
  18 import java.net.URI;
  19 import java.text.MessageFormat;
  20 import java.util.List;
  21 import javax.annotation.PostConstruct;
  22 import org.apache.commons.lang3.StringUtils;
  23 import org.onap.nbi.OnapComponentsUrlPaths;
  24 import org.onap.nbi.apis.hub.model.Event;
  25 import org.onap.nbi.apis.hub.model.EventType;
  26 import org.onap.nbi.apis.hub.model.ServiceInstanceEvent;
  27 import org.onap.nbi.apis.hub.repository.SubscriberRepository;
  28 import org.onap.nbi.apis.hub.service.EventFactory;
  29 import org.onap.nbi.apis.hub.service.NotifierService;
  30 import org.onap.nbi.apis.serviceorder.model.RelatedParty;
  31 import org.slf4j.Logger;
  32 import org.slf4j.LoggerFactory;
  33 import org.springframework.beans.factory.annotation.Autowired;
  34 import org.springframework.beans.factory.annotation.Value;
  35 import org.springframework.http.HttpEntity;
  36 import org.springframework.http.HttpHeaders;
  37 import org.springframework.http.HttpMethod;
  38 import org.springframework.http.ResponseEntity;
  39 import org.springframework.stereotype.Service;
  40 import org.springframework.util.CollectionUtils;
  41 import org.springframework.web.client.RestTemplate;
  42 import org.springframework.web.util.UriComponentsBuilder;
  43 import com.fasterxml.jackson.core.JsonParseException;
  44 import com.fasterxml.jackson.databind.JsonMappingException;
  45 import com.fasterxml.jackson.databind.JsonNode;
  46 import com.fasterxml.jackson.databind.ObjectMapper;
  47
  48 @Service
  49 public class CheckDMaaPEventsManager {
  50
  51   public static final String RESPONSE_STATUS = "response status : ";
  52   public static final String RETURNS = " returns ";
  53   public static final String ERROR_ON_CALLING = "error on calling ";
  54   private final Logger logger = LoggerFactory.getLogger(CheckDMaaPEventsManager.class);
  55   @Autowired
  56   private RestTemplate restTemplate;
  57   @Autowired
  58   private SubscriberRepository subscriberRepository;
  59   @Autowired
  60   private NotifierService notifier;
  61   @Value("${dmaap.host}")
  62   private String dmaapHostname;
  63   @Value("${dmaap.aai.topic}")
  64   private String aaiTopic;
  65   @Value("${dmaap.sdc.topic}")
  66   private String sdcTopic;
  67   @Value("${dmaap.consumergroup}")
  68   private String consumerGroup;
  69   @Value("${dmaap.consumerid}")
  70   private String consumerId;
  71   @Value("${dmaap.timeout}")
  72   private String timeout;
  73   private String dmaapGetEventsUrl;
  74
  75   @PostConstruct
  76   private void setUpAndLogDMaaPUrl() {
  77     dmaapGetEventsUrl = new StringBuilder().append(dmaapHostname)
  78         .append(OnapComponentsUrlPaths.DMAAP_CONSUME_EVENTS).toString();
  79     logger.info("DMaaP Get Events url :  " + dmaapGetEventsUrl);
  80   }
  81
  82   public void checkForDMaaPAAIEvents() {
  83     ObjectMapper mapper = new ObjectMapper();
  84     List<String> dmaapResponse = callDMaaPGetEvents(aaiTopic);
  85     if (!CollectionUtils.isEmpty(dmaapResponse)) {
  86       for (int i = 0; i < dmaapResponse.size(); i++) {
  87         String aaiEventString = dmaapResponse.get(i);
  88         if (logger.isDebugEnabled()) {
  89           logger.debug("aai event returned was {}", aaiEventString);
  90         }
  91         try {
  92           JsonNode jsonNode = mapper.readValue(aaiEventString, JsonNode.class);
  93           JsonNode eventHeader = jsonNode.get("event-header");
  94           String aaiEventEntityType = eventHeader.get("entity-type").asText();
  95           String action = eventHeader.get("action").asText();
  96           if (logger.isDebugEnabled()) {
  97             logger.debug("aaiEventEntityType is {} and action is {}", aaiEventEntityType, action);
  98           }
  99           if (aaiEventEntityType.equals("service-instance")) {
 100             {
 101               // parse the AAI-EVENT service-instance tree
 102               ServiceInstanceEvent serviceInstanceEvent = new ServiceInstanceEvent();
 103               RelatedParty relatedParty = new RelatedParty();
 104               JsonNode entity = jsonNode.get("entity");
 105               relatedParty.setId(entity.get("global-customer-id").asText());
 106               relatedParty.setName(entity.get("subscriber-name").asText());
 107               serviceInstanceEvent.setRelatedParty(relatedParty);
 108               JsonNode childServiceSubscription = entity.get("service-subscriptions");
 109               JsonNode serviceSubscriptions = childServiceSubscription.get("service-subscription");
 110               JsonNode serviceSubscription = serviceSubscriptions.get(0);
 111               String serviceSubscriptionPrint = serviceSubscription.toString();
 112               JsonNode childserviceInstances = serviceSubscription.get("service-instances");
 113               JsonNode serviceInstances = childserviceInstances.get("service-instance");
 114               JsonNode serviceInstance = serviceInstances.get(0);
 115               serviceInstanceEvent.setId(serviceInstance.get("service-instance-id").asText());
 116               serviceInstanceEvent
 117                   .setHref("service/" + serviceInstance.get("service-instance-id").asText());
 118               if (serviceInstance.get("orchestration-status") != null) {
 119                 serviceInstanceEvent.setState(serviceInstance.get("orchestration-status").asText());
 120               }
 121               if (action.equals("CREATE")) {
 122                 if (logger.isDebugEnabled()) {
 123                   logger.debug("sending service inventory event to listeners");
 124                 }
 125                 processEvent(
 126                     EventFactory.getEvent(EventType.SERVICE_CREATION, serviceInstanceEvent));
 127               } else if (action.equals("DELETE")) {
 128                 processEvent(EventFactory.getEvent(EventType.SERVICE_REMOVE, serviceInstanceEvent));
 129               } else if (action.equals("UPDATE")) {
 130                 processEvent(EventFactory.getEvent(EventType.SERVICE_ATTRIBUTE_VALUE_CHANGE,
 131                     serviceInstanceEvent));
 132               }
 133
 134             }
 135
 136           }
 137
 138         } catch (JsonParseException e) {
 139           logger.error(" unable to Parse AAI Event JSON String {}, exception is", aaiEventString,
 140               e.getMessage());
 141         } catch (JsonMappingException e) {
 142           logger.error(" unable to Map AAI Event JSON String {} to Java Pojo, exception is",
 143               aaiEventString, e.getMessage());
 144         } catch (IOException e) {
 145           logger.error("IO Error when parsing AAI Event JSON String {} ", aaiEventString,
 146               e.getMessage());
 147         }
 148       }
 149     }
 150   }
 151
 152   public void checkForDMaaPSDCEvents() {
 153     List<String> dmaapResponse = callDMaaPGetEvents(sdcTopic);
 154     if (!CollectionUtils.isEmpty(dmaapResponse)) {
 155       for (int i = 0; i < dmaapResponse.size(); i++) {
 156         String sdcEventString = dmaapResponse.get(i);
 157         if (logger.isDebugEnabled()) {
 158           logger.debug("sdc event returned was {}", sdcEventString);
 159         }
 160         processEvent(EventFactory.getEvent(EventType.SDC_DISTRIBUTION, sdcEventString));
 161       }
 162     }
 163   }
 164
 165   public List<String> callDMaaPGetEvents(String topic) {
 166
 167     URI callURI = buildRequest(topic);
 168     ResponseEntity<Object> response = callDMaaP(callURI);
 169     if (response != null) {
 170       return (List<String>) response.getBody();
 171
 172     } else {
 173       return null;
 174     }
 175   }
 176
 177   public ResponseEntity<Object> callCheckConnectivity() {
 178     URI callURI = buildRequest(null);
 179
 180     ResponseEntity<Object> response =
 181         restTemplate.exchange(callURI, HttpMethod.GET, buildRequestHeader(), Object.class);
 182
 183     if (logger.isDebugEnabled()) {
 184       logger.debug("CheckConnectivity response body : {} ", response.getBody().toString());
 185       logger.debug("CheckConnectivity response status : {}", response.getStatusCodeValue());
 186       logger.debug("CheckConnectivity callURI is : {} ", callURI.toString());
 187     }
 188     return response;
 189
 190   }
 191
 192   private URI buildRequest(String topic) {
 193     if (StringUtils.isEmpty(topic)) {
 194       topic = aaiTopic;
 195     }
 196     String dmaapGetEventsUrlFormated = dmaapGetEventsUrl.replace("$topic", topic);
 197     dmaapGetEventsUrlFormated = dmaapGetEventsUrlFormated.replace("$consumergroup", consumerGroup);
 198     dmaapGetEventsUrlFormated = dmaapGetEventsUrlFormated.replace("$consumerid", consumerId);
 199     dmaapGetEventsUrlFormated = dmaapGetEventsUrlFormated.replace("$timeout", timeout);
 200     if (logger.isDebugEnabled()) {
 201       logger.debug("Calling DMaaP Url : " + dmaapGetEventsUrlFormated);
 202     }
 203     UriComponentsBuilder callURI = UriComponentsBuilder.fromHttpUrl(dmaapGetEventsUrlFormated);
 204     return callURI.build().encode().toUri();
 205   }
 206
 207   private ResponseEntity<Object> callDMaaP(URI callURI) {
 208     try {
 209       ResponseEntity<Object> response =
 210           restTemplate.exchange(callURI, HttpMethod.GET, buildRequestHeader(), Object.class);
 211       if (logger.isDebugEnabled()) {
 212         logger.debug("response body : {} ", response.getBody().toString());
 213         logger.debug("response status : {}", response.getStatusCodeValue());
 214       }
 215       return response;
 216     } catch (Exception e) {
 217       String message = MessageFormat.format("Exception while calling dmaap URI: {0}", callURI);
 218       logger.error(message);
 219       logger.error("Exception while calling DMaaP is {} ", e.toString());
 220       return null;
 221     }
 222
 223   }
 224
 225   private HttpEntity<String> buildRequestHeader() {
 226     HttpHeaders httpHeaders = new HttpHeaders();
 227     httpHeaders.add("Accept", "application/json");
 228     httpHeaders.add("Content-Type", "application/json");
 229     return new HttpEntity<>("parameters", httpHeaders);
 230   }
 231
 232   /**
 233    * Retrieve subscribers that match an event and fire notification asynchronously
 234    */
 235   private void processEvent(Event event) {
 236     subscriberRepository.findSubscribersUsingEvent(event).forEach(sub -> notifier.run(sub, event));
 237   }
 238
 239 }