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-2018 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  * ============LICENSE_END=========================================================
  22  */
  23
  24 package org.onap.appc.client.impl.protocol;
  25
  26 import com.att.nsa.cambria.client.CambriaClientBuilders.ConsumerBuilder;
  27 import com.att.nsa.cambria.client.CambriaConsumer;
  28
  29 import java.io.IOException;
  30 import java.lang.reflect.Field;
  31 import java.net.MalformedURLException;
  32 import java.security.GeneralSecurityException;
  33 import java.util.ArrayList;
  34 import java.util.Collection;
  35 import java.util.List;
  36
  37 class ConsumerImpl implements Consumer {
  38
  39     private static final int DEFAULT_LIMIT = 1000;
  40
  41     private Collection<String> hosts;
  42     private String topic;
  43     private String group;
  44     private String groupId;
  45     private int timeout;
  46
  47     private String authKey;
  48     private String authSecret;
  49
  50     private CambriaConsumer consumer = null;
  51
  52     /**
  53      * constructor
  54      * @param urls
  55      * @param topicName
  56      * @param consumerName
  57      * @param consumerId
  58      * @param timeout
  59      */
  60     public ConsumerImpl(Collection<String> urls, String topicName, String consumerName, String consumerId, Integer timeout, String apiKey, String apiSecret) throws MalformedURLException, GeneralSecurityException, NoSuchFieldException, IllegalAccessException {
  61         this.hosts = urls;
  62         this.topic = topicName;
  63         this.group = consumerName;
  64         this.groupId = consumerId;
  65         this.authKey = apiKey;
  66         this.authSecret = apiSecret;
  67         this.timeout = timeout;
  68         consumer = getConsumer();
  69     }
  70
  71
  72     public List<String> fetch() throws IOException {
  73
  74         return fetch(DEFAULT_LIMIT);
  75     }
  76
  77     public List<String> fetch(int limit) throws IOException {
  78
  79         List<String> out = new ArrayList<String>();
  80         try {
  81             for(String msg : consumer.fetch(timeout,limit)){
  82                 out.add(msg);
  83             }
  84         } catch (IOException e) {
  85             throw e;
  86         }
  87         return out;
  88     }
  89
  90     public void registerForRead() throws IOException {
  91
  92         int waitForRegisteration = 1; //return from fetch after 1ms, no need to read any messages
  93         consumer.fetch(waitForRegisteration, 1);
  94     }
  95
  96     /**
  97      * init cambria consumer
  98      * @return CambriaConsumer
  99      */
 100     private CambriaConsumer getConsumer() throws MalformedURLException, GeneralSecurityException, NoSuchFieldException, IllegalAccessException {
 101
 102         ConsumerBuilder builder = new ConsumerBuilder();
 103
 104         builder.usingHosts(hosts).onTopic(topic).knownAs(group, groupId);
 105         builder.withSocketTimeout(timeout + 5000).waitAtServer(timeout);
 106         builder.receivingAtMost(DEFAULT_LIMIT);
 107
 108         //added for loop for defect 354330
 109         for (String url : hosts) {
 110             if (url.contains("3905") || url.contains("https")) {
 111                 builder.usingHttps(true);
 112                 break;
 113             }
 114         }
 115
 116         // Add credentials if provided
 117         if (authKey != null && authSecret != null) {
 118
 119             Field apiKeyField = ConsumerBuilder.class.getDeclaredField("fApiKey");
 120             apiKeyField.setAccessible(true);
 121             apiKeyField.set(builder, "");
 122             builder.authenticatedBy(authKey, authSecret);
 123         }
 124
 125         return builder.build();
 126     }
 127
 128     @Override
 129     public void close() {
 130         consumer.close();
 131     }
 132 }