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 (entity.contains("{")) {
  77                         // Get rid of opening ["
  78                         entity = entity.substring(2);
  79                         // Get rid of closing "]
  80                         entity = entity.substring(0, entity.length() - 2);
  81                         // This replacement effectively un-escapes the JSON
  82                         for (String message : entity.split("\",\"")) {
  83                             try {
  84                                 processMsg(message.replace("\\\"", "\""));
  85                             } catch (InvalidMessageException e) {
  86                                 Log.error("Message could not be processed", e);
  87                             }
  88                         }
  89                     } else {
  90                         Log.info("Entity doesn't appear to contain JSON elements");
  91                     }
  92                 } catch (Exception e) {
  93                     Log.error("GET " + uri + " failed.", e);
  94                 } finally {
  95                     Log.info("Pausing " + fetchPause + " milliseconds before fetching from " + uri + " again.");
  96                     try {
  97                         Thread.sleep(fetchPause);
  98                     } catch (InterruptedException e) {
  99                         Log.error("Could not sleep thread", e);
 100                         Thread.currentThread().interrupt();
 101                     }
 102                 }
 103             }
 104         }
 105     }
 106
 107     @Override
 108     public void init(Properties baseProperties, String consumerPropertiesPath) {
 109         try {
 110             baseProperties.load(new FileInputStream(new File(consumerPropertiesPath)));
 111             processProperties(baseProperties);
 112         } catch (FileNotFoundException e) {
 113             Log.error("FileNotFoundException while reading consumer properties", e);
 114         } catch (IOException e) {
 115             Log.error("IOException while reading consumer properties", e);
 116         }
 117     }
 118
 119     protected void processProperties(Properties properties) {
 120         this.properties = properties;
 121         String username = properties.getProperty("username");
 122         String password = properties.getProperty("password");
 123         String topic = properties.getProperty("topic");
 124         String group = properties.getProperty("group");
 125         String host = properties.getProperty("host");
 126         String id = properties.getProperty("id");
 127
 128         String filter = properties.getProperty("filter");
 129         if (filter != null) {
 130             if (filter.length() > 0) {
 131                 try {
 132                     filter = URLEncoder.encode(filter, StandardCharsets.UTF_8.name());
 133                 } catch (UnsupportedEncodingException e) {
 134                     Log.error("Filter could not be encoded, setting to null", e);
 135                     filter = null;
 136                 }
 137             } else {
 138                 filter = null;
 139             }
 140         }
 141
 142         String limitString = properties.getProperty("limit", DEFAULT_LIMIT);
 143         Integer limit = null;
 144         if (limitString != null && limitString.length() > 0) {
 145             limit = Integer.valueOf(limitString);
 146         }
 147
 148         Integer timeoutQueryParamValue =
 149                 Integer.valueOf(properties.getProperty("timeout", DEFAULT_TIMEOUT_QUERY_PARAM_VALUE));
 150         Integer connectTimeoutSeconds = Integer
 151                 .valueOf(properties.getProperty("connectTimeoutSeconds", DEFAULT_CONNECT_TIMEOUT_SECONDS));
 152         Integer readTimeoutMinutes =
 153                 Integer.valueOf(properties.getProperty("readTimeoutMinutes", DEFAULT_READ_TIMEOUT_MINUTES));
 154         this.client = getClient(connectTimeoutSeconds, readTimeoutMinutes);
 155         this.uri = buildUri(topic, group, id, host, timeoutQueryParamValue, limit, filter);
 156         Builder builder = client.target(uri).request("application/json");
 157         if (username != null && password != null && username.length() > 0 && password.length() > 0) {
 158             String authorizationString = buildAuthorizationString(username, password);
 159             builder.header("Authorization", authorizationString);
 160         }
 161
 162         this.getMessages = builder.buildGet();
 163         this.fetchPause = Integer.valueOf(properties.getProperty("fetchPause",DEFAULT_FETCH_PAUSE));
 164         this.isReady = true;
 165     }
 166
 167     @Override
 168     public void processMsg(String msg) throws InvalidMessageException {
 169         System.out.println(msg);
 170     }
 171
 172     @Override
 173     public boolean isReady() {
 174         return isReady;
 175     }
 176
 177     @Override
 178     public boolean isRunning() {
 179         return isRunning;
 180     }
 181
 182     protected String buildAuthorizationString(String userName, String password) {
 183         String basicAuthString = userName + ":" + password;
 184         basicAuthString = Base64.getEncoder().encodeToString(basicAuthString.getBytes());
 185         return "Basic " + basicAuthString;
 186     }
 187
 188     protected Client getClient(Integer connectTimeoutSeconds, Integer readTimeoutMinutes) {
 189         ClientBuilder clientBuilder = ClientBuilder.newBuilder();
 190         clientBuilder.connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS);
 191         clientBuilder.readTimeout(readTimeoutMinutes, TimeUnit.MINUTES);
 192         return clientBuilder.build();
 193     }
 194
 195     protected URI buildUri(String topic, String consumerGroup, String consumerId, String host, Integer timeout,
 196             Integer limit, String filter) {
 197         UriBuilder builder = UriBuilder.fromPath("http://" + host + "/events/{topic}/{consumerGroup}/{consumderId}");
 198         builder.queryParam("timeout", timeout);
 199         if (limit != null) {
 200             builder.queryParam("limit", limit);
 201         }
 202         if (filter != null) {
 203             builder.queryParam("filter", filter);
 204         }
 205         return builder.build(topic, consumerGroup, consumerId);
 206     }
 207
 208 }