PolicyEngineUtils/src/main/java/org/onap/policy/utils/BusPublisher.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * PolicyEngineUtils
   4  * ================================================================================
   5  * Copyright (C) 2017, 2019 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.utils;
  22
  23 import java.util.ArrayList;
  24 import java.util.List;
  25 import java.util.Properties;
  26 import java.util.concurrent.TimeUnit;
  27
  28 import org.onap.dmaap.mr.client.impl.MRSimplerBatchPublisher;
  29 import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants;
  30 import org.onap.policy.common.logging.eelf.PolicyLogger;
  31 import org.onap.policy.common.logging.flexlogger.FlexLogger;
  32 import org.onap.policy.common.logging.flexlogger.Logger;
  33
  34 public interface BusPublisher {
  35
  36     /**
  37      * sends a message
  38      * .
  39      * @param partitionId id
  40      * @param message the message
  41      * @return true if success, false otherwise
  42      * @throws IllegalArgumentException if no message provided
  43      */
  44     public boolean send(String partitionId, String message);
  45
  46     /**
  47      * closes the publisher.
  48      */
  49     public void close();
  50
  51     /**
  52      * DmaapClient library wrapper.
  53      */
  54     public static class DmaapPublisherWrapper implements BusPublisher {
  55         private static Logger logger = FlexLogger.getLogger(DmaapPublisherWrapper.class);
  56         /**
  57          * MR based Publisher.
  58          */
  59         protected MRSimplerBatchPublisher publisher;
  60
  61         /**
  62          * DmaapPublisherWrapper constructor.
  63          *
  64          * @param servers list of servers
  65          * @param topic topic
  66          * @param aafLogin login
  67          * @param aafPassword password
  68          */
  69         public DmaapPublisherWrapper(List<String> servers, String topic, String aafLogin, String aafPassword) {
  70
  71             ArrayList<String> dmaapServers = new ArrayList<>();
  72             for (String server : servers) {
  73                 dmaapServers.add(server + ":3904");
  74             }
  75
  76             this.publisher = new MRSimplerBatchPublisher.Builder().againstUrls(dmaapServers).onTopic(topic).build();
  77
  78             this.publisher.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
  79
  80             this.publisher.setUsername(aafLogin);
  81             this.publisher.setPassword(aafPassword);
  82
  83             Properties props = new Properties();
  84             props.setProperty("Protocol", "http");
  85             props.setProperty("contenttype", "application/json");
  86
  87             this.publisher.setProps(props);
  88
  89             this.publisher.setHost(servers.get(0));
  90
  91             if (PolicyLogger.isInfoEnabled()) {
  92                 PolicyLogger.info(DmaapPublisherWrapper.class.getName(), "CREATION: " + this);
  93             }
  94         }
  95
  96         /**
  97          * {@inheritDoc}
  98          */
  99         @Override
 100         public void close() {
 101             if (logger.isInfoEnabled()) {
 102                 logger.info(DmaapPublisherWrapper.class.getName() + "CREATION: " + this);
 103             }
 104
 105             try {
 106                 this.publisher.close(1, TimeUnit.SECONDS);
 107             } catch (Exception e) {
 108                 logger.warn(DmaapPublisherWrapper.class.getName() + "CLOSE: " + this + " because of " + e.getMessage(),
 109                         e);
 110             }
 111         }
 112
 113         /**
 114          * {@inheritDoc}
 115          */
 116         @Override
 117         public boolean send(String partitionId, String message) {
 118             if (message == null) {
 119                 throw new IllegalArgumentException("No message provided");
 120             }
 121
 122             this.publisher.send(partitionId, message);
 123             return true;
 124
 125         }
 126
 127         @Override
 128         public String toString() {
 129             StringBuilder builder = new StringBuilder();
 130             builder.append("DmaapPublisherWrapper [").append("publisher.getAuthDate()=").append(publisher.getAuthDate())
 131                     .append(", publisher.getAuthKey()=").append(publisher.getAuthKey()).append(", publisher.getHost()=")
 132                     .append(publisher.getHost()).append(", publisher.getProtocolFlag()=")
 133                     .append(publisher.getProtocolFlag()).append(", publisher.getUsername()=")
 134                     .append(publisher.getUsername()).append(", publisher.getPendingMessageCount()=")
 135                     .append(publisher.getPendingMessageCount()).append("]");
 136             return builder.toString();
 137         }
 138     }
 139
 140 }