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

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