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

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * policy-endpoints
   4  * ================================================================================
   5  * Copyright (C) 2017-2020 AT&T Intellectual Property. All rights reserved.
   6  * Modifications 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 com.att.nsa.apiClient.http.HttpClient.ConnectionType;
  25 import com.att.nsa.cambria.client.CambriaBatchingPublisher;
  26 import com.att.nsa.cambria.client.CambriaClientBuilders;
  27 import com.att.nsa.cambria.client.CambriaClientBuilders.PublisherBuilder;
  28 import java.net.MalformedURLException;
  29 import java.security.GeneralSecurityException;
  30 import java.util.ArrayList;
  31 import java.util.List;
  32 import java.util.Map;
  33 import java.util.Properties;
  34 import java.util.concurrent.TimeUnit;
  35 import org.apache.commons.lang3.StringUtils;
  36 import org.onap.dmaap.mr.client.impl.MRSimplerBatchPublisher;
  37 import org.onap.dmaap.mr.client.response.MRPublisherResponse;
  38 import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants;
  39 import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
  40 import org.onap.policy.common.gson.annotation.GsonJsonIgnore;
  41 import org.slf4j.Logger;
  42 import org.slf4j.LoggerFactory;
  43
  44 public interface BusPublisher {
  45
  46     /**
  47      * sends a message.
  48      *
  49      * @param partitionId id
  50      * @param message the message
  51      * @return true if success, false otherwise
  52      * @throws IllegalArgumentException if no message provided
  53      */
  54     public boolean send(String partitionId, String message);
  55
  56     /**
  57      * closes the publisher.
  58      */
  59     public void close();
  60
  61     /**
  62      * Cambria based library publisher.
  63      */
  64     public static class CambriaPublisherWrapper implements BusPublisher {
  65
  66         private static Logger logger = LoggerFactory.getLogger(CambriaPublisherWrapper.class);
  67
  68         /**
  69          * The actual Cambria publisher.
  70          */
  71         @GsonJsonIgnore
  72         protected CambriaBatchingPublisher publisher;
  73
  74         /**
  75          * Constructor.
  76          *
  77          * @param busTopicParams topic parameters
  78          */
  79         public CambriaPublisherWrapper(BusTopicParams busTopicParams) {
  80
  81             PublisherBuilder builder = new CambriaClientBuilders.PublisherBuilder();
  82
  83             builder.usingHosts(busTopicParams.getServers()).onTopic(busTopicParams.getTopic());
  84
  85             // Set read timeout to 30 seconds (TBD: this should be configurable)
  86             builder.withSocketTimeout(30000);
  87
  88             if (busTopicParams.isUseHttps()) {
  89                 if (busTopicParams.isAllowSelfSignedCerts()) {
  90                     builder.withConnectionType(ConnectionType.HTTPS_NO_VALIDATION);
  91                 } else {
  92                     builder.withConnectionType(ConnectionType.HTTPS);
  93                 }
  94             }
  95
  96
  97             if (busTopicParams.isApiKeyValid() && busTopicParams.isApiSecretValid()) {
  98                 builder.authenticatedBy(busTopicParams.getApiKey(), busTopicParams.getApiSecret());
  99             }
 100
 101             if (busTopicParams.isUserNameValid() && busTopicParams.isPasswordValid()) {
 102                 builder.authenticatedByHttp(busTopicParams.getUserName(), busTopicParams.getPassword());
 103             }
 104
 105             try {
 106                 this.publisher = builder.build();
 107             } catch (MalformedURLException | GeneralSecurityException e) {
 108                 throw new IllegalArgumentException(e);
 109             }
 110         }
 111
 112         @Override
 113         public boolean send(String partitionId, String message) {
 114             if (message == null) {
 115                 throw new IllegalArgumentException("No message provided");
 116             }
 117
 118             try {
 119                 this.publisher.send(partitionId, message);
 120             } catch (Exception e) {
 121                 logger.warn("{}: SEND of {} cannot be performed because of {}", this, message, e.getMessage(), e);
 122                 return false;
 123             }
 124             return true;
 125         }
 126
 127         @Override
 128         public void close() {
 129             logger.info("{}: CLOSE", this);
 130
 131             try {
 132                 this.publisher.close();
 133             } catch (Exception e) {
 134                 logger.warn("{}: CLOSE FAILED because of {}", this, e.getMessage(), e);
 135             }
 136         }
 137
 138
 139         @Override
 140         public String toString() {
 141             return "CambriaPublisherWrapper []";
 142         }
 143
 144     }
 145
 146     /**
 147      * DmaapClient library wrapper.
 148      */
 149     public abstract class DmaapPublisherWrapper implements BusPublisher {
 150
 151         private static Logger logger = LoggerFactory.getLogger(DmaapPublisherWrapper.class);
 152
 153         /**
 154          * MR based Publisher.
 155          */
 156         protected MRSimplerBatchPublisher publisher;
 157         protected Properties props;
 158
 159         /**
 160          * MR Publisher Wrapper.
 161          *
 162          * @param servers messaging bus hosts
 163          * @param topic topic
 164          * @param username AAF or DME2 Login
 165          * @param password AAF or DME2 Password
 166          */
 167         public DmaapPublisherWrapper(ProtocolTypeConstants protocol, List<String> servers, String topic,
 168                 String username, String password, boolean useHttps) {
 169
 170
 171             if (StringUtils.isBlank(topic)) {
 172                 throw new IllegalArgumentException("No topic for DMaaP");
 173             }
 174
 175
 176             configureProtocol(topic, protocol, servers, useHttps);
 177
 178             this.publisher.logTo(LoggerFactory.getLogger(MRSimplerBatchPublisher.class.getName()));
 179
 180             this.publisher.setUsername(username);
 181             this.publisher.setPassword(password);
 182
 183             props = new Properties();
 184
 185             props.setProperty("Protocol", (useHttps ? "https" : "http"));
 186             props.setProperty("contenttype", "application/json");
 187             props.setProperty("username", username);
 188             props.setProperty("password", password);
 189
 190             props.setProperty("topic", topic);
 191
 192             this.publisher.setProps(props);
 193
 194             if (protocol == ProtocolTypeConstants.AAF_AUTH) {
 195                 this.publisher.setHost(servers.get(0));
 196             }
 197
 198             logger.info("{}: CREATION: using protocol {}", this, protocol.getValue());
 199         }
 200
 201         private void configureProtocol(String topic, ProtocolTypeConstants protocol, List<String> servers,
 202                         boolean useHttps) {
 203
 204             if (protocol == ProtocolTypeConstants.AAF_AUTH) {
 205                 if (servers == null || servers.isEmpty()) {
 206                     throw new IllegalArgumentException("No DMaaP servers or DME2 partner provided");
 207                 }
 208
 209                 ArrayList<String> dmaapServers = new ArrayList<>();
 210                 String port = useHttps ? ":3905" : ":3904";
 211                 for (String server : servers) {
 212                     dmaapServers.add(server + port);
 213                 }
 214
 215
 216                 this.publisher = new MRSimplerBatchPublisher.Builder().againstUrls(dmaapServers).onTopic(topic).build();
 217
 218                 this.publisher.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
 219
 220             } else if (protocol == ProtocolTypeConstants.DME2) {
 221                 ArrayList<String> dmaapServers = new ArrayList<>();
 222                 dmaapServers.add("0.0.0.0:3904");
 223
 224                 this.publisher = new MRSimplerBatchPublisher.Builder().againstUrls(dmaapServers).onTopic(topic).build();
 225
 226                 this.publisher.setProtocolFlag(ProtocolTypeConstants.DME2.getValue());
 227
 228             } else {
 229                 throw new IllegalArgumentException("Invalid DMaaP protocol " + protocol);
 230             }
 231         }
 232
 233         @Override
 234         public void close() {
 235             logger.info("{}: CLOSE", this);
 236
 237             try {
 238                 this.publisher.close(1, TimeUnit.SECONDS);
 239             } catch (Exception e) {
 240                 logger.warn("{}: CLOSE FAILED because of {}", this, e.getMessage(), e);
 241             }
 242         }
 243
 244         @Override
 245         public boolean send(String partitionId, String message) {
 246             if (message == null) {
 247                 throw new IllegalArgumentException("No message provided");
 248             }
 249
 250             this.publisher.setPubResponse(new MRPublisherResponse());
 251             this.publisher.send(partitionId, message);
 252             MRPublisherResponse response = this.publisher.sendBatchWithResponse();
 253             if (response != null) {
 254                 logger.debug("DMaaP publisher received {} : {}", response.getResponseCode(),
 255                         response.getResponseMessage());
 256             }
 257
 258             return true;
 259         }
 260
 261         @Override
 262         public String toString() {
 263             return "DmaapPublisherWrapper [" + "publisher.getAuthDate()=" + publisher.getAuthDate()
 264                     + ", publisher.getAuthKey()=" + publisher.getAuthKey() + ", publisher.getHost()="
 265                     + publisher.getHost() + ", publisher.getProtocolFlag()=" + publisher.getProtocolFlag()
 266                     + ", publisher.getUsername()=" + publisher.getUsername() + "]";
 267         }
 268     }
 269
 270     /**
 271      * DmaapClient library wrapper.
 272      */
 273     public static class DmaapAafPublisherWrapper extends DmaapPublisherWrapper {
 274         /**
 275          * MR based Publisher.
 276          */
 277         public DmaapAafPublisherWrapper(List<String> servers, String topic, String aafLogin, String aafPassword,
 278                 boolean useHttps) {
 279
 280             super(ProtocolTypeConstants.AAF_AUTH, servers, topic, aafLogin, aafPassword, useHttps);
 281         }
 282     }
 283
 284     public static class DmaapDmePublisherWrapper extends DmaapPublisherWrapper {
 285
 286         /**
 287          * Constructor.
 288          *
 289          * @param busTopicParams topic parameters
 290          */
 291         public DmaapDmePublisherWrapper(BusTopicParams busTopicParams) {
 292
 293             super(ProtocolTypeConstants.DME2, busTopicParams.getServers(), busTopicParams.getTopic(),
 294                     busTopicParams.getUserName(), busTopicParams.getPassword(), busTopicParams.isUseHttps());
 295
 296             String dme2RouteOffer = busTopicParams.isAdditionalPropsValid()
 297                             ? busTopicParams.getAdditionalProps().get(
 298                                             PolicyEndPointProperties.DME2_ROUTE_OFFER_PROPERTY)
 299                             : null;
 300
 301             validateParams(busTopicParams, dme2RouteOffer);
 302
 303             String serviceName = busTopicParams.getServers().get(0);
 304
 305             /* These are required, no defaults */
 306             props.setProperty("Environment", busTopicParams.getEnvironment());
 307             props.setProperty("AFT_ENVIRONMENT", busTopicParams.getAftEnvironment());
 308
 309             props.setProperty(PolicyEndPointProperties.DME2_SERVICE_NAME_PROPERTY, serviceName);
 310
 311             if (busTopicParams.getPartner() != null) {
 312                 props.setProperty("Partner", busTopicParams.getPartner());
 313             }
 314             if (dme2RouteOffer != null) {
 315                 props.setProperty(PolicyEndPointProperties.DME2_ROUTE_OFFER_PROPERTY, dme2RouteOffer);
 316             }
 317
 318             props.setProperty("Latitude", busTopicParams.getLatitude());
 319             props.setProperty("Longitude", busTopicParams.getLongitude());
 320
 321             // ServiceName also a default, found in additionalProps
 322
 323             /* These are optional, will default to these values if not set in optionalProps */
 324             props.setProperty("AFT_DME2_EP_READ_TIMEOUT_MS", "50000");
 325             props.setProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", "240000");
 326             props.setProperty("AFT_DME2_EP_CONN_TIMEOUT", "15000");
 327             props.setProperty("Version", "1.0");
 328             props.setProperty("SubContextPath", "/");
 329             props.setProperty("sessionstickinessrequired", "no");
 330
 331             /* These should not change */
 332             props.setProperty("TransportType", "DME2");
 333             props.setProperty("MethodType", "POST");
 334
 335             if (busTopicParams.isAdditionalPropsValid()) {
 336                 addAdditionalProps(busTopicParams);
 337             }
 338
 339             this.publisher.setProps(props);
 340         }
 341
 342         private void validateParams(BusTopicParams busTopicParams, String dme2RouteOffer) {
 343             if (busTopicParams.isEnvironmentInvalid()) {
 344                 throw parmException(busTopicParams.getTopic(),
 345                         PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX);
 346             }
 347             if (busTopicParams.isAftEnvironmentInvalid()) {
 348                 throw parmException(busTopicParams.getTopic(),
 349                         PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX);
 350             }
 351             if (busTopicParams.isLatitudeInvalid()) {
 352                 throw parmException(busTopicParams.getTopic(),
 353                         PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LATITUDE_SUFFIX);
 354             }
 355             if (busTopicParams.isLongitudeInvalid()) {
 356                 throw parmException(busTopicParams.getTopic(),
 357                         PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LONGITUDE_SUFFIX);
 358             }
 359
 360             if ((busTopicParams.isPartnerInvalid())
 361                     && StringUtils.isBlank(dme2RouteOffer)) {
 362                 throw new IllegalArgumentException(
 363                         "Must provide at least " + PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
 364                                 + busTopicParams.getTopic()
 365                                 + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_PARTNER_SUFFIX + " or "
 366                                 + PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "." + busTopicParams.getTopic()
 367                                 + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX + " for DME2");
 368             }
 369         }
 370
 371         private void addAdditionalProps(BusTopicParams busTopicParams) {
 372             for (Map.Entry<String, String> entry : busTopicParams.getAdditionalProps().entrySet()) {
 373                 String key = entry.getKey();
 374                 String value = entry.getValue();
 375
 376                 if (value != null) {
 377                     props.setProperty(key, value);
 378                 }
 379             }
 380         }
 381
 382         private IllegalArgumentException parmException(String topic, String propnm) {
 383             return new IllegalArgumentException("Missing " + PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
 384                     + topic + propnm + " property for DME2 in DMaaP");
 385
 386         }
 387     }
 388 }