dmaap-listener/src/main/java/org/onap/ccsdk/sli/northbound/dmaapclient/MessageRouterHttpClient.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * openECOMP : SDN-C
   4  * ================================================================================
   5  * Copyright (C) 2017 - 2018 AT&T Intellectual Property. All rights
   6  *                                              reserved.
   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.ccsdk.sli.northbound.dmaapclient;
  23
  24 import java.io.File;
  25 import java.io.FileInputStream;
  26 import java.io.FileNotFoundException;
  27 import java.io.IOException;
  28 import java.io.UnsupportedEncodingException;
  29 import java.net.URI;
  30 import java.net.URLEncoder;
  31 import java.nio.charset.StandardCharsets;
  32 import java.util.Base64;
  33 import java.util.Properties;
  34 import java.util.concurrent.TimeUnit;
  35 import javax.ws.rs.client.Client;
  36 import javax.ws.rs.client.ClientBuilder;
  37 import javax.ws.rs.client.Invocation;
  38 import javax.ws.rs.client.Invocation.Builder;
  39 import javax.ws.rs.core.Response;
  40 import javax.ws.rs.core.UriBuilder;
  41 import org.slf4j.Logger;
  42 import org.slf4j.LoggerFactory;
  43
  44 /*
  45  * jax-rs based client to build message router consumers
  46  */
  47 public class MessageRouterHttpClient implements SdncDmaapConsumer {
  48     private static final Logger Log = LoggerFactory.getLogger(MessageRouterHttpClient.class);
  49
  50     protected Boolean isReady = false;
  51     protected Boolean isRunning = false;
  52     protected Client client;
  53     protected URI uri;
  54     protected Invocation getMessages;
  55     protected Integer fetchPause;
  56     protected Properties properties;
  57     protected final String DEFAULT_CONNECT_TIMEOUT_SECONDS = "30";
  58     protected final String DEFAULT_READ_TIMEOUT_MINUTES = "3";
  59     protected final String DEFAULT_TIMEOUT_QUERY_PARAM_VALUE = "15000";
  60     protected final String DEFAULT_LIMIT = null;
  61     protected final String DEFAULT_FETCH_PAUSE = "5000";
  62
  63     public MessageRouterHttpClient() {
  64
  65     }
  66
  67         @Override
  68         public void run() {
  69                 if (isReady) {
  70                         isRunning = true;
  71                         while (isRunning) {
  72                                 try {
  73                                         Response response = getMessages.invoke();
  74                                         Log.info("GET " + uri + " returned http status " + response.getStatus());
  75                                         String entity = response.readEntity(String.class);
  76                                         if (response.getStatus() < 300) {
  77                                                 if (entity.contains("{")) {
  78                                                         // Get rid of opening ["
  79                                                         entity = entity.substring(2);
  80                                                         // Get rid of closing "]
  81                                                         entity = entity.substring(0, entity.length() - 2);
  82                                                         // This replacement effectively un-escapes the JSON
  83                                                         for (String message : entity.split("\",\"")) {
  84                                                                 try {
  85                                                                         processMsg(message.replace("\\\"", "\""));
  86                                                                 } catch (InvalidMessageException e) {
  87                                                                         Log.error("Message could not be processed", e);
  88                                                                 }
  89                                                         }
  90                                                 } else {
  91                                                         if (entity.length() < 1) {
  92                                                                 Log.info("GET was successful, but the server returned an empty message body.");
  93                                                         } else {
  94                                                                 Log.info(
  95                                                                                 "GET was successful, but entity is not valid JSON. Message body will be logged, but not processed");
  96                                                                 Log.info(entity);
  97                                                         }
  98                                                 }
  99                                         } else {
 100                                                 Log.info("GET failed, message body will be logged, but not processed.");
 101                                                 Log.info(entity);
 102                                         }
 103                                 } catch (Exception e) {
 104                                         Log.error("GET " + uri + " failed.", e);
 105                                 } finally {
 106                                         Log.info("Pausing " + fetchPause + " milliseconds before fetching from " + uri + " again.");
 107                                         try {
 108                                                 Thread.sleep(fetchPause);
 109                                         } catch (InterruptedException e) {
 110                                                 Log.error("Could not sleep thread", e);
 111                                                 Thread.currentThread().interrupt();
 112                                         }
 113                                 }
 114                         }
 115                 }
 116         }
 117
 118     @Override
 119     public void init(Properties baseProperties, String consumerPropertiesPath) {
 120         try {
 121             baseProperties.load(new FileInputStream(new File(consumerPropertiesPath)));
 122             processProperties(baseProperties);
 123         } catch (FileNotFoundException e) {
 124             Log.error("FileNotFoundException while reading consumer properties", e);
 125         } catch (IOException e) {
 126             Log.error("IOException while reading consumer properties", e);
 127         }
 128     }
 129
 130     protected void processProperties(Properties properties) {
 131         this.properties = properties;
 132         String username = properties.getProperty("username");
 133         String password = properties.getProperty("password");
 134         String topic = properties.getProperty("topic");
 135         String group = properties.getProperty("group");
 136         String host = properties.getProperty("host");
 137         String id = properties.getProperty("id");
 138
 139         String filter = properties.getProperty("filter");
 140         if (filter != null) {
 141             if (filter.length() > 0) {
 142                 try {
 143                     filter = URLEncoder.encode(filter, StandardCharsets.UTF_8.name());
 144                 } catch (UnsupportedEncodingException e) {
 145                     Log.error("Filter could not be encoded, setting to null", e);
 146                     filter = null;
 147                 }
 148             } else {
 149                 filter = null;
 150             }
 151         }
 152
 153         String limitString = properties.getProperty("limit", DEFAULT_LIMIT);
 154         Integer limit = null;
 155         if (limitString != null && limitString.length() > 0) {
 156             limit = Integer.valueOf(limitString);
 157         }
 158
 159         Integer timeoutQueryParamValue =
 160                 Integer.valueOf(properties.getProperty("timeout", DEFAULT_TIMEOUT_QUERY_PARAM_VALUE));
 161         Integer connectTimeoutSeconds = Integer
 162                 .valueOf(properties.getProperty("connectTimeoutSeconds", DEFAULT_CONNECT_TIMEOUT_SECONDS));
 163         Integer readTimeoutMinutes =
 164                 Integer.valueOf(properties.getProperty("readTimeoutMinutes", DEFAULT_READ_TIMEOUT_MINUTES));
 165         this.client = getClient(connectTimeoutSeconds, readTimeoutMinutes);
 166         this.uri = buildUri(topic, group, id, host, timeoutQueryParamValue, limit, filter);
 167         Builder builder = client.target(uri).request("application/json");
 168         if (username != null && password != null && username.length() > 0 && password.length() > 0) {
 169             String authorizationString = buildAuthorizationString(username, password);
 170             builder.header("Authorization", authorizationString);
 171         }
 172
 173         this.getMessages = builder.buildGet();
 174         this.fetchPause = Integer.valueOf(properties.getProperty("fetchPause",DEFAULT_FETCH_PAUSE));
 175         this.isReady = true;
 176     }
 177
 178     @Override
 179     public void processMsg(String msg) throws InvalidMessageException {
 180         System.out.println(msg);
 181     }
 182
 183     @Override
 184     public boolean isReady() {
 185         return isReady;
 186     }
 187
 188     @Override
 189     public boolean isRunning() {
 190         return isRunning;
 191     }
 192
 193     protected String buildAuthorizationString(String userName, String password) {
 194         String basicAuthString = userName + ":" + password;
 195         basicAuthString = Base64.getEncoder().encodeToString(basicAuthString.getBytes());
 196         return "Basic " + basicAuthString;
 197     }
 198
 199     protected Client getClient(Integer connectTimeoutSeconds, Integer readTimeoutMinutes) {
 200         ClientBuilder clientBuilder = ClientBuilder.newBuilder();
 201         clientBuilder.connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS);
 202         clientBuilder.readTimeout(readTimeoutMinutes, TimeUnit.MINUTES);
 203         return clientBuilder.build();
 204     }
 205
 206     protected URI buildUri(String topic, String consumerGroup, String consumerId, String host, Integer timeout,
 207             Integer limit, String filter) {
 208         UriBuilder builder = UriBuilder.fromPath("http://" + host + "/events/{topic}/{consumerGroup}/{consumderId}");
 209         builder.queryParam("timeout", timeout);
 210         if (limit != null) {
 211             builder.queryParam("limit", limit);
 212         }
 213         if (filter != null) {
 214             builder.queryParam("filter", filter);
 215         }
 216         return builder.build(topic, consumerGroup, consumerId);
 217     }
 218
 219 }