gerrit.onap Code Review - policy/drools-pdp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * policy-endpoints
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.openecomp.policy.drools.event.comm.bus.internal;
  22
  23 import java.util.List;
  24 import java.util.Map;
  25
  26 import org.openecomp.policy.drools.event.comm.Topic;
  27 import org.openecomp.policy.drools.event.comm.bus.DmaapTopicSource;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30
  31 /**
  32  * This topic reader implementation specializes in reading messages
  33  * over DMAAP topic and notifying its listeners
  34  */
  35 public class SingleThreadedDmaapTopicSource extends SingleThreadedBusTopicSource
  36                                             implements DmaapTopicSource, Runnable {
  37
  38         private static Logger logger = LoggerFactory.getLogger(SingleThreadedDmaapTopicSource.class);
  39
  40         protected boolean allowSelfSignedCerts;
  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 servers DMaaP servers
  56          * @param topic DMaaP Topic to be monitored
  57          * @param apiKey DMaaP API Key (optional)
  58          * @param apiSecret DMaaP API Secret (optional)
  59          * @param consumerGroup DMaaP Reader Consumer Group
  60          * @param consumerInstance DMaaP Reader Instance
  61          * @param fetchTimeout DMaaP fetch timeout
  62          * @param fetchLimit DMaaP fetch limit
  63          * @param environment DME2 Environment
  64          * @param aftEnvironment DME2 AFT Environment
  65          * @param partner DME2 Partner
  66          * @param latitude DME2 Latitude
  67          * @param longitude DME2 Longitude
  68          * @param additionalProps Additional properties to pass to DME2
  69          * @param useHttps does connection use HTTPS?
  70          * @param allowSelfSignedCerts are self-signed certificates allow
  71          *
  72          * @throws IllegalArgumentException An invalid parameter passed in
  73          */
  74         public SingleThreadedDmaapTopicSource(List<String> servers, String topic,
  75                                                                                         String apiKey, String apiSecret,
  76                                                                                         String userName, String password,
  77                                                                                         String consumerGroup, String consumerInstance,
  78                                                                                         int fetchTimeout, int fetchLimit,
  79                                                                                         String environment, String aftEnvironment, String partner,
  80                                                                                         String latitude, String longitude, Map<String,String> additionalProps,
  81                                                                                         boolean useHttps, boolean allowSelfSignedCerts)
  82                         throws IllegalArgumentException {
  83
  84                 super(servers, topic, apiKey, apiSecret,
  85                           consumerGroup, consumerInstance,
  86                           fetchTimeout, fetchLimit, useHttps,allowSelfSignedCerts);
  87
  88                 this.userName = userName;
  89                 this.password = password;
  90
  91                 this.environment = environment;
  92                 this.aftEnvironment = aftEnvironment;
  93                 this.partner = partner;
  94
  95                 this.latitude = latitude;
  96                 this.longitude = longitude;
  97
  98                 this.additionalProps = additionalProps;
  99                 try {
 100                         this.init();
 101                 } catch (Exception e) {
 102                         logger.error("ERROR during init of topic {}", this.topic);
 103                         throw new IllegalArgumentException(e);
 104                 }
 105         }
 106
 107         /**
 108          *
 109          * @param servers DMaaP servers
 110          * @param topic DMaaP Topic to be monitored
 111          * @param apiKey DMaaP API Key (optional)
 112          * @param apiSecret DMaaP API Secret (optional)
 113          * @param consumerGroup DMaaP Reader Consumer Group
 114          * @param consumerInstance DMaaP Reader Instance
 115          * @param fetchTimeout DMaaP fetch timeout
 116          * @param fetchLimit DMaaP fetch limit
 117          * @param useHttps does connection use HTTPS?
 118          * @param allowSelfSignedCerts are self-signed certificates allow
 119          * @throws IllegalArgumentException An invalid parameter passed in
 120          */
 121         public SingleThreadedDmaapTopicSource(List<String> servers, String topic,
 122                                                       String apiKey, String apiSecret,
 123                                                       String userName, String password,
 124                                                       String consumerGroup, String consumerInstance,
 125                                                       int fetchTimeout, int fetchLimit, boolean useHttps, boolean allowSelfSignedCerts)
 126                         throws IllegalArgumentException {
 127
 128
 129                 super(servers, topic, apiKey, apiSecret,
 130                           consumerGroup, consumerInstance,
 131                           fetchTimeout, fetchLimit, useHttps, allowSelfSignedCerts);
 132
 133                 this.userName = userName;
 134                 this.password = password;
 135
 136                 try {
 137                         this.init();
 138                 } catch (Exception e) {
 139                         logger.warn("dmaap-source: cannot create topic {} because of {}", topic, e.getMessage(), e);
 140                         throw new IllegalArgumentException(e);
 141                 }
 142         }
 143
 144
 145         /**
 146          * Initialize the Cambria or MR Client
 147          */
 148         @Override
 149         public void init() throws Exception {
 150                 if (this.userName == null || this.userName.isEmpty() ||
 151                                 this.password == null || this.password.isEmpty()) {
 152                                 this.consumer =
 153                                                 new BusConsumer.CambriaConsumerWrapper(this.servers, this.topic,
 154                                                                                            this.apiKey, this.apiSecret,
 155                                                                                            this.consumerGroup, this.consumerInstance,
 156                                                                                            this.fetchTimeout, this.fetchLimit,
 157                                                                                            this.useHttps, this.allowSelfSignedCerts);
 158                 } else if ((this.environment == null    || this.environment.isEmpty()) &&
 159                                    (this.aftEnvironment == null || this.aftEnvironment.isEmpty()) &&
 160                                    (this.latitude == null          || this.latitude.isEmpty()) &&
 161                                    (this.longitude == null         || this.longitude.isEmpty()) &&
 162                                    (this.partner == null           || this.partner.isEmpty())) {
 163                         this.consumer =
 164                                         new BusConsumer.DmaapAafConsumerWrapper(this.servers, this.topic,
 165                                                                                     this.apiKey, this.apiSecret,
 166                                                                                     this.userName, this.password,
 167                                                                                     this.consumerGroup, this.consumerInstance,
 168                                                                                     this.fetchTimeout, this.fetchLimit, this.useHttps);
 169                 } else {
 170                         this.consumer =
 171                                         new BusConsumer.DmaapDmeConsumerWrapper(this.servers, this.topic,
 172                                                                                     this.apiKey, this.apiSecret,
 173                                                                                     this.userName, this.password,
 174                                                                                     this.consumerGroup, this.consumerInstance,
 175                                                                                     this.fetchTimeout, this.fetchLimit,
 176                                                                                     this.environment, this.aftEnvironment, this.partner,
 177                                                                                     this.latitude, this.longitude, this.additionalProps, this.useHttps);
 178                 }
 179
 180                 logger.info("{}: INITTED", this);
 181         }
 182
 183         /**
 184          * {@inheritDoc}
 185          */
 186         @Override
 187         public CommInfrastructure getTopicCommInfrastructure() {
 188                 return Topic.CommInfrastructure.DMAAP;
 189         }
 190
 191         @Override
 192         public String toString() {
 193                 StringBuilder builder = new StringBuilder();
 194                 builder.append("SingleThreadedDmaapTopicSource [userName=").append(userName).append(", password=")
 195                                 .append((password == null || password.isEmpty()) ? "-" : password.length())
 196                                 .append(", getTopicCommInfrastructure()=").append(getTopicCommInfrastructure())
 197                                 .append(", toString()=").append(super.toString()).append("]");
 198                 return builder.toString();
 199         }
 200
 201
 202 }