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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * PolicyEngineUtils
   4  * ================================================================================
   5  * Copyright (C) 2017, 2019 AT&T Intellectual Property. All rights reserved.
   6  * Modifications copyright (c) 2019 Nokia
   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.utils;
  23
  24 import java.net.MalformedURLException;
  25 import java.util.List;
  26 import java.util.Properties;
  27
  28 import org.onap.dmaap.mr.client.MRClient.MRApiException;
  29 import org.onap.dmaap.mr.client.impl.MRConsumerImpl;
  30 import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants;
  31
  32 public interface BusConsumer {
  33
  34     /**
  35      * fetch messages.
  36      *
  37      * @return list of messages
  38      * @throws MRApiException when error encountered by underlying libraries
  39      */
  40     Iterable<String> fetch() throws MRApiException;
  41
  42     /**
  43      * close underlying library consumer.
  44      */
  45     void close();
  46
  47     /**
  48      * MR based consumer.
  49      */
  50     class DmaapConsumerWrapper implements BusConsumer {
  51
  52         /**
  53          * MR Consumer.
  54          */
  55         protected MRConsumerImpl consumer;
  56
  57         /**
  58          * MR Consumer Wrapper.
  59          *
  60          * @param servers messaging bus hosts
  61          * @param topic topic
  62          * @param apiKey API Key
  63          * @param apiSecret API Secret
  64          * @param aafLogin AAF Login
  65          * @param aafPassword AAF Password
  66          * @param consumerGroup Consumer Group
  67          * @param consumerInstance Consumer Instance
  68          * @param fetchTimeout Fetch Timeout
  69          * @param fetchLimit Fetch Limit
  70          */
  71         public DmaapConsumerWrapper(List<String> servers, String topic, String aafLogin, String aafPassword,
  72                 String consumerGroup, String consumerInstance, int fetchTimeout, int fetchLimit)
  73                 throws MalformedURLException {
  74
  75             this(new MRConsumerImpl(servers, topic, consumerGroup, consumerInstance, fetchTimeout, fetchLimit, null,
  76                     aafLogin, aafPassword), aafLogin, aafPassword, servers.get(0));
  77
  78         }
  79
  80         DmaapConsumerWrapper(MRConsumerImpl consumer, String aafLogin, String aafPassword, String host) {
  81             this.consumer = consumer;
  82             this.consumer.setUsername(aafLogin);
  83             this.consumer.setPassword(aafPassword);
  84             this.consumer.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
  85             this.consumer.setHost(host + ":3904");
  86
  87             Properties props = new Properties();
  88             props.setProperty("Protocol", "http");
  89             this.consumer.setProps(props);
  90         }
  91
  92         /**
  93          * {@inheritDoc}
  94          */
  95         @Override
  96         public Iterable<String> fetch() throws MRApiException {
  97             try {
  98                 return consumer.fetch();
  99             } catch (Exception e) {
 100                 throw new MRApiException("Error during MR consumer Fetch ", e);
 101             }
 102         }
 103
 104         /**
 105          * {@inheritDoc}
 106          */
 107         @Override
 108         public void close() {
 109             this.consumer.close();
 110         }
 111
 112         @Override
 113         public String toString() {
 114             StringBuilder builder = new StringBuilder();
 115             builder.append("DmaapConsumerWrapper [").append("consumer.getAuthDate()=").append(consumer.getAuthDate())
 116                     .append(", consumer.getAuthKey()=").append(consumer.getAuthKey()).append(", consumer.getHost()=")
 117                     .append(consumer.getHost()).append(", consumer.getProtocolFlag()=")
 118                     .append(consumer.getProtocolFlag()).append(", consumer.getUsername()=")
 119                     .append(consumer.getUsername()).append("]");
 120             return builder.toString();
 121         }
 122     }
 123 }