appc-client/client-lib/src/main/java/org/onap/appc/client/impl/protocol/ConsumerImpl.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP : APPC
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Copyright (C) 2017 Amdocs
   8  * =============================================================================
   9  * Licensed under the Apache License, Version 2.0 (the "License");
  10  * you may not use this file except in compliance with the License.
  11  * You may obtain a copy of the License at
  12  *
  13  *      http://www.apache.org/licenses/LICENSE-2.0
  14  *
  15  * Unless required by applicable law or agreed to in writing, software
  16  * distributed under the License is distributed on an "AS IS" BASIS,
  17  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18  * See the License for the specific language governing permissions and
  19  * limitations under the License.
  20  *
  21  * ECOMP is a trademark and service mark of AT&T Intellectual Property.
  22  * ============LICENSE_END=========================================================
  23  */
  24
  25 package org.onap.appc.client.impl.protocol;
  26
  27 import com.att.nsa.cambria.client.CambriaClientBuilders.ConsumerBuilder;
  28 import com.att.nsa.cambria.client.CambriaConsumer;
  29
  30 import java.io.IOException;
  31 import java.lang.reflect.Field;
  32 import java.net.MalformedURLException;
  33 import java.security.GeneralSecurityException;
  34 import java.util.ArrayList;
  35 import java.util.Collection;
  36 import java.util.List;
  37
  38 class ConsumerImpl implements Consumer {
  39
  40     private static final int DEFAULT_LIMIT = 1000;
  41
  42     private Collection<String> hosts;
  43     private String topic;
  44     private String group;
  45     private String groupId;
  46     private int timeout;
  47
  48     private String authKey;
  49     private String authSecret;
  50
  51     private CambriaConsumer consumer = null;
  52
  53     /**
  54      * constructor
  55      * @param urls
  56      * @param topicName
  57      * @param consumerName
  58      * @param consumerId
  59      * @param timeout
  60      */
  61     public ConsumerImpl(Collection<String> urls, String topicName, String consumerName, String consumerId, Integer timeout, String apiKey, String apiSecret) throws MalformedURLException, GeneralSecurityException, NoSuchFieldException, IllegalAccessException {
  62         this.hosts = urls;
  63         this.topic = topicName;
  64         this.group = consumerName;
  65         this.groupId = consumerId;
  66         this.authKey = apiKey;
  67         this.authSecret = apiSecret;
  68         this.timeout = timeout;
  69         consumer = getConsumer();
  70     }
  71
  72
  73     public List<String> fetch() throws IOException {
  74
  75         return fetch(DEFAULT_LIMIT);
  76     }
  77
  78     public List<String> fetch(int limit) throws IOException {
  79
  80         List<String> out = new ArrayList<String>();
  81         try {
  82             for(String msg : consumer.fetch(timeout,limit)){
  83                 out.add(msg);
  84             }
  85         } catch (IOException e) {
  86             throw e;
  87         }
  88         return out;
  89     }
  90
  91     public void registerForRead() throws IOException {
  92
  93         int waitForRegisteration = 1; //return from fetch after 1ms, no need to read any messages
  94         consumer.fetch(waitForRegisteration, 1);
  95     }
  96
  97     /**
  98      * init cambria consumer
  99      * @return CambriaConsumer
 100      */
 101     private CambriaConsumer getConsumer() throws MalformedURLException, GeneralSecurityException, NoSuchFieldException, IllegalAccessException {
 102
 103         ConsumerBuilder builder = new ConsumerBuilder();
 104
 105         builder.usingHosts(hosts).onTopic(topic).knownAs(group, groupId);
 106         builder.withSocketTimeout(timeout + 5000).waitAtServer(timeout);
 107         builder.receivingAtMost(DEFAULT_LIMIT);
 108
 109         //added for loop for defect 354330
 110         for (String url : hosts) {
 111             if (url.contains("3905") || url.contains("https")) {
 112                 builder.usingHttps(true);
 113                 break;
 114             }
 115         }
 116
 117         // Add credentials if provided
 118         if (authKey != null && authSecret != null) {
 119
 120             Field apiKeyField = ConsumerBuilder.class.getDeclaredField("fApiKey");
 121             apiKeyField.setAccessible(true);
 122             apiKeyField.set(builder, "");
 123             builder.authenticatedBy(authKey, authSecret);
 124         }
 125
 126         return builder.build();
 127     }
 128
 129     @Override
 130     public void close() {
 131         consumer.close();
 132     }
 133 }