appc-adapters/appc-dmaap-adapter/appc-dmaap-adapter-bundle/src/main/java/org/onap/appc/adapter/messaging/dmaap/impl/DmaapConsumerImpl.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 java.io.IOException;
  28 import java.util.*;
  29
  30 import com.att.eelf.configuration.EELFLogger;
  31 import com.att.eelf.configuration.EELFManager;
  32 //import com.att.nsa.cambria.client.CambriaClientBuilders;
  33 //import com.att.nsa.cambria.client.CambriaClientBuilders.ConsumerBuilder;
  34 //import com.att.nsa.cambria.client.CambriaConsumer;
  35
  36 import com.att.nsa.mr.client.MRClientFactory;
  37 import com.att.nsa.mr.client.MRConsumer;
  38 import org.apache.commons.lang3.StringUtils;
  39 import org.onap.appc.adapter.message.Consumer;
  40 import org.onap.appc.configuration.Configuration;
  41 import org.onap.appc.configuration.ConfigurationFactory;
  42 import org.onap.appc.metricservice.MetricRegistry;
  43 import org.onap.appc.metricservice.MetricService;
  44 import org.onap.appc.metricservice.impl.MetricServiceImpl;
  45 import org.onap.appc.metricservice.metric.Metric;
  46 import org.onap.appc.metricservice.metric.MetricType;
  47 import org.onap.appc.metricservice.metric.DmaapRequestCounterMetric;
  48 import org.onap.appc.metricservice.policy.PublishingPolicy;
  49 import org.onap.appc.metricservice.publisher.LogPublisher;
  50 import org.osgi.framework.BundleContext;
  51 import org.osgi.framework.FrameworkUtil;
  52 import org.osgi.framework.ServiceReference;
  53
  54 public class DmaapConsumerImpl implements Consumer {
  55
  56     private final EELFLogger LOG = EELFManager.getInstance().getLogger(DmaapConsumerImpl.class);
  57     private final Configuration configuration = ConfigurationFactory.getConfiguration();
  58     // Default values
  59     private static final int DEFAULT_TIMEOUT_MS = 60000;
  60     private static final int DEFAULT_LIMIT = 1000;
  61     private static MetricRegistry metricRegistry;
  62     private String topic;
  63     private boolean isMetricEnabled=false;
  64     private boolean useHttps = false;
  65     private MRConsumer client = null;
  66     private Properties props = null;
  67
  68
  69     public DmaapConsumerImpl(Collection<String> urls, String topicName, String consumerGroupName, String consumerId,String user, String password) {
  70         this(urls, topicName, consumerGroupName, consumerId,user, password,null);
  71
  72     }
  73
  74     public DmaapConsumerImpl(Collection<String> urls, String topicName, String consumerGroupName, String consumerId,String user, String password,String filter) {
  75         this.topic = topicName;
  76         this.props = new Properties();
  77         String urlsStr = StringUtils.join(urls, ',');
  78         props.setProperty("host",urlsStr);
  79         props.setProperty("group",consumerGroupName);
  80         props.setProperty("id",consumerId);
  81         props.setProperty("username",user);
  82         props.setProperty("password",password);
  83         if(filter != null) {
  84             props.setProperty("filter", filter);
  85         }
  86     }
  87
  88
  89     private void initMetric() {
  90         LOG.debug("Metric getting initialized");
  91         MetricService metricService = getMetricservice();
  92         metricRegistry = metricService.createRegistry("APPC");
  93         DmaapRequestCounterMetric dmaapKpiMetric = metricRegistry.metricBuilderFactory().
  94                 dmaapRequestCounterBuilder().
  95                 withName("DMAAP_KPI").withType(MetricType.COUNTER).
  96                 withRecievedMessage(0)
  97                 .withPublishedMessage(0)
  98                 .build();
  99         if (metricRegistry.register(dmaapKpiMetric)) {
 100             Metric[] metrics = new Metric[]{dmaapKpiMetric};
 101             LogPublisher logPublisher = new LogPublisher(metricRegistry, metrics);
 102             LogPublisher[] logPublishers = new LogPublisher[1];
 103             logPublishers[0] = logPublisher;
 104             PublishingPolicy manuallyScheduledPublishingPolicy = metricRegistry.policyBuilderFactory().
 105                     scheduledPolicyBuilder().withPublishers(logPublishers).
 106                     withMetrics(metrics).
 107                     build();
 108             LOG.debug("Policy getting initialized");
 109             manuallyScheduledPublishingPolicy.init();
 110             LOG.debug("Metric initialized");
 111         }
 112     }
 113
 114
 115     /**
 116      * @return An instance of MRConsumer created from our class variables
 117      */
 118     private synchronized MRConsumer getClient(int waitMs, int limit) {
 119         try {
 120             props.setProperty("timeout",String.valueOf(waitMs));
 121             props.setProperty("limit",String.valueOf(limit));
 122             String topicProducerPropFileName = DmaapUtil.createConsumerPropFile(topic,props);
 123             return MRClientFactory.createConsumer ( topicProducerPropFileName);
 124         } catch (IOException e1) {
 125             LOG.error("failed to createConsumer",e1);
 126             return null;
 127         }
 128     }
 129
 130     @Override
 131     public synchronized void updateCredentials(String key, String secret) {
 132         LOG.info(String.format("Setting auth to %s for %s", key, this.toString()));
 133         props.setProperty("user",String.valueOf(key));
 134         props.setProperty("password",String.valueOf(secret));
 135         client = null;
 136     }
 137
 138     @Override
 139     public List<String> fetch(int waitMs, int limit) {
 140         Properties properties=configuration.getProperties();
 141         if(properties!=null && properties.getProperty("metric.enabled")!=null ){
 142           isMetricEnabled=Boolean.valueOf(properties.getProperty("metric.enabled"));
 143         }
 144         if(isMetricEnabled){
 145             initMetric();
 146         }
 147         LOG.debug(String.format("Fetching up to %d records with %dms wait on %s", limit, waitMs, this.toString()));
 148         List<String> out = new ArrayList<>();
 149
 150         // Create client once and reuse it on subsequent fetches. This is
 151         // to support failover to other servers in the DMaaP cluster.
 152         if (client == null) {
 153             LOG.info("Getting DMaaP Client ...");
 154             client = getClient(waitMs, limit);
 155         }
 156         try {
 157             for (String s : client.fetch(waitMs, limit)) {
 158                 out.add(s);
 159                 if(isMetricEnabled){
 160                     ((DmaapRequestCounterMetric)metricRegistry.metric("DMAAP_KPI")).incrementRecievedMessage();
 161                 }
 162             }
 163             LOG.debug(String.format("Got %d records from %s", out.size(), this.toString()));
 164         } catch (Exception e) {
 165             // Connection exception
 166             LOG.error(String.format("Dmaap Connection Issue Detected. %s", e.getMessage()));
 167             e.printStackTrace();
 168             try {
 169                 LOG.warn(String.format("Sleeping for %dms to compensate for connection failure", waitMs));
 170                 Thread.sleep(waitMs);
 171             } catch (InterruptedException e2) {
 172                 LOG.warn(String.format("Failed to wait for %dms after bad fetch", waitMs));
 173             }
 174         }
 175
 176
 177         return out;
 178     }
 179
 180     /**
 181      * Close consumer Dmaap client
 182      */
 183     @Override
 184     public void close() {
 185         LOG.debug("Closing Dmaap consumer client....");
 186         if (client != null) {
 187             client.close();
 188         }
 189     }
 190
 191     @Override
 192     public List<String> fetch() {
 193         return fetch(DEFAULT_TIMEOUT_MS, DEFAULT_LIMIT);
 194     }
 195
 196     @Override
 197     public String toString() {
 198         String hostStr = (props == null || props.getProperty("host") == null? "N/A" : props.getProperty("host"));
 199         String group = (props == null || props.getProperty("group") == null? "N/A" : props.getProperty("group"));
 200         String id = (props == null || props.getProperty("id") == null? "N/A" : props.getProperty("id"));
 201         return String.format("Consumer %s/%s listening to %s on [%s]", group, id, topic, hostStr);
 202     }
 203
 204     @Override
 205     public void useHttps(boolean yes) {
 206         useHttps = yes;
 207     }
 208
 209
 210     private MetricService getMetricservice() {
 211         BundleContext bctx = FrameworkUtil.getBundle(MetricService.class).getBundleContext();
 212         // Get AAIadapter reference
 213         ServiceReference sref = bctx.getServiceReference(MetricService.class.getName());
 214         if (sref != null) {
 215             LOG.info("Metric Service from bundlecontext");
 216             return (MetricServiceImpl) bctx.getService(sref);
 217
 218         } else {
 219             LOG.info("Metric Service error from bundlecontext");
 220             LOG.warn("Cannot find service reference for org.onap.appc.metricservice.MetricService");
 221             return null;
 222
 223         }
 224     }
 225
 226
 227 }