appc-oam/appc-oam-bundle/src/main/java/org/onap/appc/oam/messageadapter/MessageAdapter.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP : APPC
   4  * ================================================================================
   5  * Copyright (C) 2017-2018 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  * ============LICENSE_END=========================================================
  22  */
  23
  24 package org.onap.appc.oam.messageadapter;
  25
  26 import com.att.eelf.configuration.EELFLogger;
  27 import com.att.eelf.configuration.EELFManager;
  28 import com.fasterxml.jackson.core.JsonProcessingException;
  29 import org.apache.commons.lang.ObjectUtils;
  30 import org.onap.appc.adapter.message.MessageAdapterFactory;
  31 import org.onap.appc.adapter.message.Producer;
  32 import org.onap.appc.configuration.Configuration;
  33 import org.onap.appc.configuration.ConfigurationFactory;
  34 import org.osgi.framework.BundleContext;
  35 import org.osgi.framework.FrameworkUtil;
  36 import org.osgi.framework.ServiceReference;
  37
  38 import java.util.HashSet;
  39 import java.util.Properties;
  40
  41 public class MessageAdapter {
  42
  43     private final EELFLogger logger = EELFManager.getInstance().getLogger(MessageAdapter.class);
  44
  45     private final String PROP_APPC_OAM_DISABLED = "appc.OAM.disabled";
  46     private final String PROP_APPC_OAM_TOPIC_WRITE = "appc.OAM.topic.write";
  47     private String PROP_APPC_OAM_CLIENT_KEY = "appc.OAM.client.key";
  48     private String PROP_APPC_OAM_CLIENT_SECRET = "appc.OAM.client.secret";
  49     private String PROP_APPC_OAM_POOLMEMBERS = "appc.OAM.poolMembers";
  50
  51     private Producer producer;
  52     private String partition;
  53     private Configuration configuration;
  54     private HashSet<String> pool;
  55     private String writeTopic;
  56     private String apiKey;
  57     private String apiSecret;
  58     private boolean isDisabled;
  59
  60     /**
  61      * Initialize producer client to post messages using configuration properties.
  62      */
  63     public void init() {
  64         configuration = ConfigurationFactory.getConfiguration();
  65         Properties properties = configuration.getProperties();
  66         updateProperties(properties);
  67
  68         if (isAppcOamPropsListenerEnabled()) {
  69             createProducer();
  70         } else {
  71             logger.warn(String.format("The listener %s is disabled and will not be run", "appc.OAM"));
  72         }
  73     }
  74
  75     /**
  76      * Create producer using MessageAdapterFactory which is found through bundle context.
  77      */
  78     void createProducer() {
  79         BundleContext ctx = FrameworkUtil.getBundle(MessageAdapter.class).getBundleContext();
  80         if (ctx == null) {
  81             logger.warn("MessageAdapter cannot create producer due to no bundle context.");
  82             return;
  83         }
  84
  85         ServiceReference svcRef = ctx.getServiceReference(MessageAdapterFactory.class.getName());
  86         if (svcRef == null) {
  87             logger.warn("MessageAdapter cannot create producer due to no MessageAdapterFactory service reference.");
  88             return;
  89         }
  90
  91         Producer localProducer = ((MessageAdapterFactory) ctx.getService(svcRef)).createProducer(pool, writeTopic,
  92                 apiKey, apiSecret);
  93
  94         for (String url : pool) {
  95             if (url.contains("3905") || url.contains("https")) {
  96                 localProducer.useHttps(true);
  97                 break;
  98             }
  99         }
 100
 101         producer = localProducer;
 102
 103         logger.debug("MessageAdapter created producer.");
 104     }
 105
 106     /**
 107      * Read property value to set writeTopic, apiKey, apiSecret and pool.
 108      *
 109      * @param props of configuration
 110      */
 111     private void updateProperties(Properties props) {
 112         logger.trace("Entering to updateProperties with Properties = " + ObjectUtils.toString(props));
 113
 114         pool = new HashSet<>();
 115         if (props != null) {
 116             isDisabled = Boolean.parseBoolean(props.getProperty(PROP_APPC_OAM_DISABLED));
 117             writeTopic = props.getProperty(PROP_APPC_OAM_TOPIC_WRITE);
 118             apiKey = props.getProperty(PROP_APPC_OAM_CLIENT_KEY);
 119             apiSecret = props.getProperty(PROP_APPC_OAM_CLIENT_SECRET);
 120             String hostnames = props.getProperty(PROP_APPC_OAM_POOLMEMBERS);
 121             if (hostnames != null && !hostnames.isEmpty()) {
 122                 for (String name : hostnames.split(",")) {
 123                     pool.add(name);
 124                 }
 125             }
 126         }
 127     }
 128
 129     /**
 130      * Get producer. If it is null, call createProducer to create it again.
 131      *
 132      * @return Producer
 133      */
 134     Producer getProducer() {
 135         if (producer == null) {
 136             // In case, producer was not properly set yet, set it again.
 137             logger.info("Calling createProducer as producer is null.");
 138             createProducer();
 139         }
 140
 141         return producer;
 142     }
 143
 144     /**
 145      * Posts message to UEB. As UEB accepts only json messages this method first convert uebMessage to json format
 146      * and post it to UEB.
 147      *
 148      * @param oamContext response data that based on it a message will be send to UEB (the format of the message that
 149      *                   will be sent to UEB based on the action and its YANG domainmodel).
 150      */
 151     public void post(OAMContext oamContext) {
 152         if (logger.isTraceEnabled()) {
 153             logger.trace("Entering to post with AsyncResponse = " + ObjectUtils.toString(oamContext));
 154         }
 155
 156         boolean success;
 157         String jsonMessage;
 158         try {
 159             jsonMessage = Converter.convAsyncResponseToUebOutgoingMessageJsonString(oamContext);
 160             if (logger.isDebugEnabled()) {
 161                 logger.debug("UEB Response = " + jsonMessage);
 162             }
 163
 164             Producer myProducer = getProducer();
 165             success = myProducer != null && myProducer.post(this.partition, jsonMessage);
 166         } catch (JsonProcessingException e1) {
 167             logger.error("Error generating Json from UEB message " + e1.getMessage());
 168             success = false;
 169         } catch (Exception e) {
 170             logger.error("Error sending message to UEB " + e.getMessage(), e);
 171             success = false;
 172         }
 173
 174         if (logger.isTraceEnabled()) {
 175             logger.trace("Exiting from post with (success = " + ObjectUtils.toString(success) + ")");
 176         }
 177     }
 178
 179     private boolean isAppcOamPropsListenerEnabled() {
 180         return !isDisabled;
 181     }
 182 }