gerrit.onap Code Review - dcaegen2/collectors/datafile.git/blob

   1 /*
   2  * ============LICENSE_START======================================================================
   3  * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
   4  * ===============================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   6  * in compliance with the License. You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software distributed under the License
  11  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  12  * or implied. See the License for the specific language governing permissions and limitations under
  13  * the License.
  14  * ============LICENSE_END========================================================================
  15  */
  16
  17 package org.onap.dcaegen2.collectors.datafile.service.producer;
  18
  19 import com.google.gson.JsonElement;
  20 import com.google.gson.JsonParser;
  21
  22 import java.io.IOException;
  23 import java.io.InputStream;
  24 import java.net.URI;
  25 import java.nio.charset.StandardCharsets;
  26 import java.util.concurrent.Future;
  27
  28 import javax.net.ssl.SSLContext;
  29
  30 import org.apache.commons.codec.binary.Base64;
  31 import org.apache.commons.io.IOUtils;
  32 import org.apache.http.HttpResponse;
  33 import org.apache.http.client.methods.HttpPut;
  34 import org.apache.http.conn.ssl.NoopHostnameVerifier;
  35 import org.apache.http.entity.ByteArrayEntity;
  36 import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
  37 import org.apache.http.impl.nio.client.HttpAsyncClients;
  38 import org.apache.http.ssl.SSLContextBuilder;
  39
  40 import org.onap.dcaegen2.collectors.datafile.io.FileSystemResourceWrapper;
  41 import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
  42 import org.onap.dcaegen2.collectors.datafile.model.CommonFunctions;
  43 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
  44 import org.onap.dcaegen2.collectors.datafile.service.HttpUtils;
  45 import org.onap.dcaegen2.collectors.datafile.web.PublishRedirectStrategy;
  46 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49 import org.springframework.http.HttpHeaders;
  50 import org.springframework.web.util.DefaultUriBuilderFactory;
  51
  52 import reactor.core.publisher.Flux;
  53
  54 /**
  55  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/4/18
  56  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  57  */
  58 public class DmaapProducerReactiveHttpClient {
  59
  60     private static final String X_ATT_DR_META = "X-ATT-DR-META";
  61     private static final String NAME_JSON_TAG = "name";
  62     private static final String INTERNAL_LOCATION_JSON_TAG = "internalLocation";
  63     private static final String URI_SEPARATOR = "/";
  64     private static final String DEFAULT_FEED_ID = "1";
  65
  66     private final Logger logger = LoggerFactory.getLogger(this.getClass());
  67
  68     private final String dmaapHostName;
  69     private final Integer dmaapPortNumber;
  70     private final String dmaapTopicName;
  71     private final String dmaapProtocol;
  72     private final String dmaapContentType;
  73     private final String user;
  74     private final String pwd;
  75
  76     private IFileSystemResource fileResource;
  77     private CloseableHttpAsyncClient webClient;
  78
  79     /**
  80      * Constructor DmaapProducerReactiveHttpClient.
  81      *
  82      * @param dmaapPublisherConfiguration - DMaaP producer configuration object
  83      */
  84     public DmaapProducerReactiveHttpClient(DmaapPublisherConfiguration dmaapPublisherConfiguration) {
  85         this.dmaapHostName = dmaapPublisherConfiguration.dmaapHostName();
  86         this.dmaapPortNumber = dmaapPublisherConfiguration.dmaapPortNumber();
  87         this.dmaapTopicName = dmaapPublisherConfiguration.dmaapTopicName();
  88         this.dmaapProtocol = dmaapPublisherConfiguration.dmaapProtocol();
  89         this.dmaapContentType = dmaapPublisherConfiguration.dmaapContentType();
  90         this.user = dmaapPublisherConfiguration.dmaapUserName();
  91         this.pwd = dmaapPublisherConfiguration.dmaapUserPassword();
  92     }
  93
  94     /**
  95      * Function for calling DMaaP HTTP producer - post request to DMaaP DataRouter.
  96      *
  97      * @param consumerDmaapModel - object which will be sent to DMaaP DataRouter
  98      * @return status code of operation
  99      */
 100     public Flux<String> getDmaapProducerResponse(ConsumerDmaapModel consumerDmaapModel) {
 101         logger.trace("Entering getDmaapProducerResponse with {}", consumerDmaapModel);
 102         try {
 103             logger.trace("Starting to publish to DR");
 104
 105             webClient = getWebClient();
 106             webClient.start();
 107
 108             HttpPut put = new HttpPut();
 109             prepareHead(consumerDmaapModel, put);
 110             prepareBody(consumerDmaapModel, put);
 111             addUserCredentialsToHead(put);
 112
 113             Future<HttpResponse> future = webClient.execute(put, null);
 114             HttpResponse response = future.get();
 115             logger.trace(response.toString());
 116             webClient.close();
 117             handleHttpResponse(response);
 118             return Flux.just(response.toString());
 119         } catch (Exception e) {
 120             logger.error("Unable to send file to DataRouter. Data: {}", consumerDmaapModel, e);
 121             return Flux.empty();
 122         }
 123     }
 124
 125     private void handleHttpResponse(HttpResponse response) {
 126         int statusCode = response.getStatusLine().getStatusCode();
 127         if (HttpUtils.isSuccessfulResponseCode(statusCode)) {
 128             logger.trace("Publish to DR successful!");
 129         } else {
 130             logger.error("Publish to DR unsuccessful, response code: " + statusCode);
 131         }
 132     }
 133
 134     private void addUserCredentialsToHead(HttpPut put) {
 135         String plainCreds = user + ":" + pwd;
 136         byte[] plainCredsBytes = plainCreds.getBytes(StandardCharsets.ISO_8859_1);
 137         byte[] base64CredsBytes = Base64.encodeBase64(plainCredsBytes);
 138         String base64Creds = new String(base64CredsBytes);
 139         logger.trace("base64Creds...: {}", base64Creds);
 140         put.addHeader("Authorization", "Basic " + base64Creds);
 141     }
 142
 143     private void prepareHead(ConsumerDmaapModel model, HttpPut put) {
 144         put.addHeader(HttpHeaders.CONTENT_TYPE, dmaapContentType);
 145         JsonElement metaData = new JsonParser().parse(new CommonFunctions().createJsonBody(model));
 146         String name = metaData.getAsJsonObject().remove(NAME_JSON_TAG).getAsString();
 147         metaData.getAsJsonObject().remove(INTERNAL_LOCATION_JSON_TAG);
 148         put.addHeader(X_ATT_DR_META, metaData.toString());
 149         put.setURI(getUri(name));
 150     }
 151
 152     private void prepareBody(ConsumerDmaapModel model, HttpPut put) {
 153         String fileLocation = model.getInternalLocation();
 154         IFileSystemResource fileSystemResource = getFileSystemResource();
 155         fileSystemResource.setPath(fileLocation);
 156         InputStream fileInputStream = null;
 157         try {
 158             fileInputStream = fileSystemResource.getInputStream();
 159         } catch (IOException e) {
 160             logger.error("Unable to get stream from filesystem.", e);
 161         }
 162         try {
 163             put.setEntity(new ByteArrayEntity(IOUtils.toByteArray(fileInputStream)));
 164         } catch (IOException e) {
 165             logger.error("Unable to set put request body from ByteArray.", e);
 166         }
 167     }
 168
 169     private URI getUri(String fileName) {
 170         String path = dmaapTopicName + URI_SEPARATOR + DEFAULT_FEED_ID + URI_SEPARATOR + fileName;
 171         return new DefaultUriBuilderFactory().builder().scheme(dmaapProtocol).host(dmaapHostName).port(dmaapPortNumber)
 172                 .path(path).build();
 173     }
 174
 175     private IFileSystemResource getFileSystemResource() {
 176         if (fileResource == null) {
 177             fileResource = new FileSystemResourceWrapper();
 178         }
 179         return fileResource;
 180     }
 181
 182     protected void setFileSystemResource(IFileSystemResource fileSystemResource) {
 183         fileResource = fileSystemResource;
 184     }
 185
 186     protected CloseableHttpAsyncClient getWebClient() {
 187         if (webClient != null) {
 188             return webClient;
 189         }
 190         SSLContext sslContext = null;
 191         try {
 192             sslContext = new SSLContextBuilder().loadTrustMaterial(null, (certificate, authType) -> true).build();
 193         } catch (Exception e) {
 194             logger.trace("Unable to get sslContext.", e);
 195         }
 196         //@formatter:off
 197         return HttpAsyncClients.custom()
 198                 .setSSLContext(sslContext)
 199                 .setSSLHostnameVerifier(new NoopHostnameVerifier())
 200                 .setRedirectStrategy(PublishRedirectStrategy.INSTANCE)
 201                 .build();
 202         //@formatter:on
 203     }
 204
 205     protected void setWebClient(CloseableHttpAsyncClient client) {
 206         this.webClient = client;
 207     }
 208 }