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

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