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