appc-adapters/appc-dmaap-adapter/appc-dmaap-adapter-bundle/src/main/java/org/openecomp/appc/adapter/dmaap/impl/EventSenderImpl.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * openECOMP : APP-C
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights
   6  *                                              reserved.
   7  * ================================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  *      http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.openecomp.appc.adapter.dmaap.impl;
  23
  24 import java.util.*;
  25 import java.util.concurrent.ConcurrentHashMap;
  26
  27 import org.openecomp.appc.adapter.dmaap.EventSender;
  28 import org.openecomp.appc.adapter.dmaap.Producer;
  29 import org.openecomp.appc.adapter.dmaap.DmaapDestination;
  30 import org.openecomp.appc.adapter.dmaap.event.EventHeader;
  31 import org.openecomp.appc.adapter.dmaap.event.EventMessage;
  32 import org.openecomp.appc.adapter.dmaap.event.EventStatus;
  33 import org.openecomp.appc.adapter.dmaap.DmaapProducer;
  34 import org.openecomp.appc.configuration.Configuration;
  35 import org.openecomp.appc.configuration.ConfigurationFactory;
  36 import org.openecomp.appc.exceptions.APPCException;
  37 import com.att.eelf.configuration.EELFLogger;
  38 import com.att.eelf.configuration.EELFManager;
  39 import org.openecomp.sdnc.sli.SvcLogicContext;
  40
  41
  42 public class EventSenderImpl implements EventSender
  43 {
  44     private static final EELFLogger LOG = EELFManager.getInstance().getLogger(EventSenderImpl.class);
  45     public static final String EVENT_TOPIC_WRITE = "event.topic.write";
  46     public static final String EVENT_CLIENT_KEY = "event.client.key";
  47     public static final String EVENT_CLIENT_SECRET = "event.client.secret";
  48     public static final String EVENT_POOL_MEMBERS = "event.pool.members";
  49
  50     private static Configuration configuration = ConfigurationFactory.getConfiguration();
  51
  52     private Map<String,Producer> producerMap = new ConcurrentHashMap<>();
  53
  54     public Map<String, Producer> getProducerMap() {
  55         return producerMap;
  56     }
  57
  58     public void setProducerMap(Map<String, Producer> producerMap) {
  59         this.producerMap = producerMap;
  60     }
  61
  62     public EventSenderImpl(){
  63
  64     }
  65
  66     public void initialize(){
  67         Properties properties = configuration.getProperties();
  68         String writeTopic;
  69         String apiKey;
  70         String apiSecret;
  71         final List<String> pool = new ArrayList<>();
  72
  73         for(DmaapDestination destination:DmaapDestination.values()){
  74             writeTopic = properties.getProperty(destination + "." +  EVENT_TOPIC_WRITE);
  75             apiKey = properties.getProperty(destination + "." + EVENT_CLIENT_KEY);
  76             apiSecret = properties.getProperty(destination + "." + EVENT_CLIENT_SECRET);
  77             String hostNames = properties.getProperty(destination + "." + EVENT_POOL_MEMBERS);
  78
  79             if (hostNames != null && !hostNames.isEmpty()) {
  80                 LOG.debug(String.format("hostNames = %s, taken from property: %s", hostNames, destination + "." + EVENT_POOL_MEMBERS));
  81                 Collections.addAll(pool, hostNames.split(","));
  82             }
  83
  84             LOG.debug(String.format("pool = %s, taken from property: %s", pool, destination + "." + EVENT_POOL_MEMBERS));
  85             LOG.debug(String.format("writeTopic = %s, taken from property: %s", writeTopic, destination + "." + EVENT_TOPIC_WRITE));
  86             LOG.debug(String.format("apiKey = %s, taken from property: %s", apiKey, destination + "." + EVENT_CLIENT_KEY));
  87             Producer producer = new DmaapProducer(pool, writeTopic);
  88
  89             if (apiKey != null && apiSecret != null) {
  90                 producer.updateCredentials(apiKey, apiSecret);
  91             }
  92
  93             for (String url : pool) {
  94                 if (url.contains("3905") || url.contains("https")) {
  95                     LOG.debug("Producer should use HTTPS");
  96                     producer.useHttps(true);
  97                     break;
  98                 }
  99             }
 100             producerMap.put(destination.toString(),producer);
 101         }
 102
 103     }
 104
 105     @Override
 106     public boolean sendEvent(DmaapDestination destination,EventMessage msg) {
 107         String jsonStr = msg.toJson();
 108         String id = msg.getEventHeader().getEventId();
 109         LOG.info(String.format("Posting Message [%s - %s]", id, jsonStr));
 110         Producer producer = producerMap.get(destination.toString());
 111         return producer.post(id, jsonStr);
 112     }
 113
 114     @Override
 115     public boolean sendEvent(DmaapDestination destination,Map<String, String> params, SvcLogicContext ctx) throws APPCException {
 116
 117         if (params == null) {
 118             String message = "Parameters map is empty (null)";
 119             LOG.error(message);
 120             throw new APPCException(message);
 121         }
 122         String eventTime = new Date(System.currentTimeMillis()).toString();
 123         String apiVer = params.get("apiVer");
 124         String eventId = params.get("eventId");
 125         String reason = params.get("reason");
 126         String entityId=params.get("entityId");
 127         if(entityId!=null){
 128             reason=reason+"("+entityId+")";
 129         }
 130         Integer code = Integer.getInteger(params.get("code"), 500);
 131
 132         if (eventTime == null || apiVer == null || eventId == null || reason == null) {
 133             String message = String.format("Missing input parameters: %s", params);
 134             LOG.error(message);
 135             throw new APPCException(message);
 136         }
 137         EventMessage dmaapEventMessage = new EventMessage(
 138                         new EventHeader(eventTime, apiVer, eventId),
 139                         new EventStatus(code, reason));
 140
 141         return sendEvent(destination,dmaapEventMessage);
 142     }
 143 }