appc-adapters/appc-dmaap-adapter/appc-dmaap-adapter-bundle/src/main/java/org/onap/appc/adapter/messaging/dmaap/impl/EventSenderDmaapImpl.java

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