gerrit.onap Code Review - policy/common.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * policy-endpoints
   4  * ================================================================================
   5  * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
   6  * Modified Copyright (C) 2018 Samsung Electronics Co., Ltd.
   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.onap.policy.common.endpoints.event.comm.bus.internal;
  23
  24 import java.net.MalformedURLException;
  25 import java.util.Map;
  26
  27 import org.onap.policy.common.endpoints.event.comm.Topic;
  28 import org.onap.policy.common.endpoints.event.comm.bus.DmaapTopicSource;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 /**
  33  * This topic reader implementation specializes in reading messages over DMAAP topic and notifying
  34  * its listeners
  35  */
  36 public class SingleThreadedDmaapTopicSource extends SingleThreadedBusTopicSource implements DmaapTopicSource, Runnable {
  37
  38     private static Logger logger = LoggerFactory.getLogger(SingleThreadedDmaapTopicSource.class);
  39
  40
  41     protected final String userName;
  42     protected final String password;
  43
  44     protected String environment = null;
  45     protected String aftEnvironment = null;
  46     protected String partner = null;
  47     protected String latitude = null;
  48     protected String longitude = null;
  49
  50     protected Map<String, String> additionalProps = null;
  51
  52
  53     /**
  54      *
  55      * @param busTopicParams Parameters object containing all the required inputs     *
  56      * @throws IllegalArgumentException An invalid parameter passed in
  57      */
  58     public SingleThreadedDmaapTopicSource(BusTopicParams busTopicParams) {
  59
  60         super(busTopicParams);
  61
  62         this.userName = busTopicParams.getUserName();
  63         this.password = busTopicParams.getPassword();
  64
  65         this.environment = busTopicParams.getEnvironment();
  66         this.aftEnvironment = busTopicParams.getAftEnvironment();
  67         this.partner = busTopicParams.getPartner();
  68
  69         this.latitude = busTopicParams.getLatitude();
  70         this.longitude = busTopicParams.getLongitude();
  71
  72         this.additionalProps = busTopicParams.getAdditionalProps();
  73         try {
  74             this.init();
  75         } catch (Exception e) {
  76             logger.error("ERROR during init in dmaap-source: cannot create topic {} because of {}", topic, e.getMessage(), e);
  77             throw new IllegalArgumentException(e);
  78         }
  79     }
  80
  81
  82     /**
  83      * Initialize the Cambria or MR Client
  84      */
  85     @Override
  86     public void init() throws MalformedURLException {
  87         if (anyNullOrEmpty(this.userName, this.password)) {
  88             this.consumer = new BusConsumer.CambriaConsumerWrapper(BusTopicParams.builder()
  89                     .servers(this.servers)
  90                     .topic(this.topic)
  91                     .apiKey(this.apiKey)
  92                     .apiSecret(this.apiSecret)
  93                     .consumerGroup(this.consumerGroup)
  94                     .consumerInstance(this.consumerInstance)
  95                     .fetchTimeout(this.fetchTimeout)
  96                     .fetchLimit(this.fetchLimit)
  97                     .useHttps(this.useHttps)
  98                     .allowSelfSignedCerts(this.allowSelfSignedCerts)
  99                     .build());
 100         } else if (allNullOrEmpty(this.environment, this.aftEnvironment, this.latitude, this.longitude, this.partner)) {
 101             this.consumer = new BusConsumer.CambriaConsumerWrapper(BusTopicParams.builder()
 102                     .servers(this.servers)
 103                     .topic(this.topic)
 104                     .apiKey(this.apiKey)
 105                     .apiSecret(this.apiSecret)
 106                     .userName(this.userName)
 107                     .password(this.password)
 108                     .consumerGroup(this.consumerGroup)
 109                     .consumerInstance(this.consumerInstance)
 110                     .fetchTimeout(this.fetchTimeout)
 111                     .fetchLimit(this.fetchLimit)
 112                     .useHttps(this.useHttps)
 113                     .allowSelfSignedCerts(this.allowSelfSignedCerts)
 114                     .build());
 115         } else {
 116             this.consumer = new BusConsumer.DmaapDmeConsumerWrapper(BusTopicParams.builder()
 117                     .servers(this.servers)
 118                     .topic(this.topic)
 119                     .apiKey(this.apiKey)
 120                     .apiSecret(this.apiSecret)
 121                     .userName(this.userName)
 122                     .password(this.password)
 123                     .consumerGroup(this.consumerGroup)
 124                     .consumerInstance(this.consumerInstance)
 125                     .fetchTimeout(this.fetchTimeout)
 126                     .fetchLimit(this.fetchLimit)
 127                     .environment(this.environment)
 128                     .aftEnvironment(this.aftEnvironment)
 129                     .partner(this.partner)
 130                     .latitude(this.latitude)
 131                     .longitude(this.longitude)
 132                     .additionalProps(this.additionalProps)
 133                     .useHttps(this.useHttps).build());
 134         }
 135
 136         logger.info("{}: INITTED", this);
 137     }
 138
 139     /**
 140      * {@inheritDoc}
 141      */
 142     @Override
 143     public CommInfrastructure getTopicCommInfrastructure() {
 144         return Topic.CommInfrastructure.DMAAP;
 145     }
 146
 147     @Override
 148     public String toString() {
 149         StringBuilder builder = new StringBuilder();
 150         builder.append("SingleThreadedDmaapTopicSource [userName=").append(userName).append(", password=")
 151                 .append((password == null || password.isEmpty()) ? "-" : password.length())
 152                 .append(", getTopicCommInfrastructure()=").append(getTopicCommInfrastructure()).append(", toString()=")
 153                 .append(super.toString()).append("]");
 154         return builder.toString();
 155     }
 156
 157
 158 }