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  * ================================================================================
   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.onap.policy.common.endpoints.event.comm.bus.internal.impl;
  22
  23 import com.att.nsa.apiClient.http.HttpClient.ConnectionType;
  24 import com.att.nsa.cambria.client.CambriaBatchingPublisher;
  25 import com.att.nsa.cambria.client.CambriaClientBuilders;
  26 import com.att.nsa.cambria.client.CambriaClientBuilders.PublisherBuilder;
  27 import com.fasterxml.jackson.annotation.JsonIgnore;
  28
  29 import java.net.MalformedURLException;
  30 import java.security.GeneralSecurityException;
  31 import java.util.List;
  32
  33 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusPublisher;
  34 import org.slf4j.Logger;
  35 import org.slf4j.LoggerFactory;
  36
  37 /**
  38  * Cambria based library publisher
  39  */
  40 public class CambriaPublisherWrapper implements BusPublisher {
  41
  42     private static Logger logger = LoggerFactory.getLogger(CambriaPublisherWrapper.class);
  43
  44     /**
  45      * The actual Cambria publisher
  46      */
  47     @JsonIgnore
  48     protected volatile CambriaBatchingPublisher publisher;
  49
  50     public CambriaPublisherWrapper(List<String> servers, String topic, String apiKey, String apiSecret,
  51             boolean useHttps) {
  52         this(servers, topic, apiKey, apiSecret, null, null, useHttps, false);
  53     }
  54
  55     public CambriaPublisherWrapper(List<String> servers, String topic, String apiKey, String apiSecret, String username,
  56             String password, boolean useHttps, boolean selfSignedCerts) {
  57
  58         PublisherBuilder builder = new CambriaClientBuilders.PublisherBuilder();
  59
  60         builder.usingHosts(servers).onTopic(topic);
  61
  62         // Set read timeout to 30 seconds (TBD: this should be configurable)
  63         builder.withSocketTimeout(30000);
  64
  65         if (useHttps) {
  66             if (selfSignedCerts) {
  67                 builder.withConnectionType(ConnectionType.HTTPS_NO_VALIDATION);
  68             } else {
  69                 builder.withConnectionType(ConnectionType.HTTPS);
  70             }
  71         }
  72
  73
  74         if (apiKey != null && !apiKey.isEmpty() && apiSecret != null && !apiSecret.isEmpty()) {
  75             builder.authenticatedBy(apiKey, apiSecret);
  76         }
  77
  78         if (username != null && !username.isEmpty() && password != null && !password.isEmpty()) {
  79             builder.authenticatedByHttp(username, password);
  80         }
  81
  82         try {
  83             this.publisher = builder.build();
  84         } catch (MalformedURLException | GeneralSecurityException e) {
  85             throw new IllegalArgumentException(e);
  86         }
  87     }
  88
  89     /**
  90      * {@inheritDoc}
  91      */
  92     @Override
  93     public boolean send(String partitionId, String message) {
  94         if (message == null) {
  95             throw new IllegalArgumentException("No message provided");
  96         }
  97
  98         try {
  99             this.publisher.send(partitionId, message);
 100         } catch (Exception e) {
 101             logger.warn("{}: SEND of {} cannot be performed because of {}", this, message, e.getMessage(), e);
 102             return false;
 103         }
 104         return true;
 105     }
 106
 107     /**
 108      * {@inheritDoc}
 109      */
 110     @Override
 111     public void close() {
 112         logger.info("{}: CLOSE", this);
 113
 114         try {
 115             this.publisher.close();
 116         } catch (Exception e) {
 117             logger.warn("{}: CLOSE FAILED because of {}", this, e.getMessage(), e);
 118         }
 119     }
 120
 121
 122     @Override
 123     public String toString() {
 124         return "CambriaPublisherWrapper []";
 125     }
 126
 127 }