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

   1 /*
   2  * ============LICENSE_START======================================================================
   3  * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 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.nio.file.Path;
  27 import java.nio.file.Paths;
  28 import java.security.KeyManagementException;
  29 import java.security.KeyStoreException;
  30 import java.security.NoSuchAlgorithmException;
  31 import java.util.concurrent.Future;
  32
  33 import javax.net.ssl.SSLContext;
  34
  35 import org.apache.commons.codec.binary.Base64;
  36 import org.apache.commons.io.IOUtils;
  37 import org.apache.http.HttpResponse;
  38 import org.apache.http.client.methods.HttpPut;
  39 import org.apache.http.conn.ssl.NoopHostnameVerifier;
  40 import org.apache.http.entity.ByteArrayEntity;
  41 import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
  42 import org.apache.http.impl.nio.client.HttpAsyncClients;
  43 import org.apache.http.ssl.SSLContextBuilder;
  44 import org.onap.dcaegen2.collectors.datafile.io.FileSystemResourceWrapper;
  45 import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
  46 import org.onap.dcaegen2.collectors.datafile.model.CommonFunctions;
  47 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
  48 import org.onap.dcaegen2.collectors.datafile.web.PublishRedirectStrategy;
  49 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
  50 import org.slf4j.Logger;
  51 import org.slf4j.LoggerFactory;
  52 import org.springframework.http.HttpHeaders;
  53 import org.springframework.http.HttpStatus;
  54 import org.springframework.web.util.DefaultUriBuilderFactory;
  55
  56 import reactor.core.publisher.Flux;
  57
  58 /**
  59  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/4/18
  60  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  61  */
  62 public class DmaapProducerReactiveHttpClient {
  63
  64     private static final String X_DMAAP_DR_META = "X-DMAAP-DR-META";
  65     private static final String NAME_JSON_TAG = "name";
  66     private static final String INTERNAL_LOCATION_JSON_TAG = "internalLocation";
  67     private static final String URI_SEPARATOR = "/";
  68     private static final String DEFAULT_FEED_ID = "1";
  69
  70     private final Logger logger = LoggerFactory.getLogger(this.getClass());
  71
  72     private final String dmaapHostName;
  73     private final Integer dmaapPortNumber;
  74     private final String dmaapTopicName;
  75     private final String dmaapProtocol;
  76     private final String dmaapContentType;
  77     private final String user;
  78     private final String pwd;
  79
  80     private IFileSystemResource fileResource = new FileSystemResourceWrapper();
  81     private CloseableHttpAsyncClient webClient;
  82
  83     /**
  84      * Constructor DmaapProducerReactiveHttpClient.
  85      *
  86      * @param dmaapPublisherConfiguration - DMaaP producer configuration object
  87      */
  88     public DmaapProducerReactiveHttpClient(DmaapPublisherConfiguration dmaapPublisherConfiguration) {
  89         this.dmaapHostName = dmaapPublisherConfiguration.dmaapHostName();
  90         this.dmaapPortNumber = dmaapPublisherConfiguration.dmaapPortNumber();
  91         this.dmaapTopicName = dmaapPublisherConfiguration.dmaapTopicName();
  92         this.dmaapProtocol = dmaapPublisherConfiguration.dmaapProtocol();
  93         this.dmaapContentType = dmaapPublisherConfiguration.dmaapContentType();
  94         this.user = dmaapPublisherConfiguration.dmaapUserName();
  95         this.pwd = dmaapPublisherConfiguration.dmaapUserPassword();
  96     }
  97
  98     /**
  99      * Function for calling DMaaP HTTP producer - post request to DMaaP DataRouter.
 100      *
 101      * @param consumerDmaapModel - object which will be sent to DMaaP DataRouter
 102      * @return status code of operation
 103      */
 104     public Flux<HttpStatus> getDmaapProducerResponse(ConsumerDmaapModel consumerDmaapModel) {
 105         logger.trace("Entering getDmaapProducerResponse with {}", consumerDmaapModel);
 106         try {
 107             logger.trace("Starting to publish to DR {}",  consumerDmaapModel.getInternalLocation());
 108
 109             webClient = getWebClient();
 110             webClient.start();
 111
 112             HttpPut put = new HttpPut();
 113             prepareHead(consumerDmaapModel, put);
 114             prepareBody(consumerDmaapModel, put);
 115             addUserCredentialsToHead(put);
 116
 117             Future<HttpResponse> future = webClient.execute(put, null);
 118             HttpResponse response = future.get();
 119             logger.trace(response.toString());
 120             webClient.close();
 121             return Flux.just(HttpStatus.valueOf(response.getStatusLine().getStatusCode()));
 122         } catch (Exception e) {
 123             logger.error("Unable to send file to DataRouter. Data: {}", consumerDmaapModel.getInternalLocation(), e);
 124             return Flux.error(e);
 125         }
 126     }
 127
 128     private void addUserCredentialsToHead(HttpPut put) {
 129         String plainCreds = user + ":" + pwd;
 130         byte[] plainCredsBytes = plainCreds.getBytes(StandardCharsets.ISO_8859_1);
 131         byte[] base64CredsBytes = Base64.encodeBase64(plainCredsBytes);
 132         String base64Creds = new String(base64CredsBytes);
 133         logger.trace("base64Creds...: {}", base64Creds);
 134         put.addHeader("Authorization", "Basic " + base64Creds);
 135     }
 136
 137     private void prepareHead(ConsumerDmaapModel model, HttpPut put) {
 138         put.addHeader(HttpHeaders.CONTENT_TYPE, dmaapContentType);
 139         JsonElement metaData = new JsonParser().parse(CommonFunctions.createJsonBody(model));
 140         String name = metaData.getAsJsonObject().remove(NAME_JSON_TAG).getAsString();
 141         metaData.getAsJsonObject().remove(INTERNAL_LOCATION_JSON_TAG);
 142         put.addHeader(X_DMAAP_DR_META, metaData.toString());
 143         put.setURI(getUri(name));
 144     }
 145
 146     private void prepareBody(ConsumerDmaapModel model, HttpPut put) throws IOException {
 147         Path fileLocation = Paths.get(model.getInternalLocation());
 148         this.fileResource.setPath(fileLocation);
 149         InputStream fileInputStream = fileResource.getInputStream();
 150
 151         put.setEntity(new ByteArrayEntity(IOUtils.toByteArray(fileInputStream)));
 152
 153     }
 154
 155     private URI getUri(String fileName) {
 156         String path = dmaapTopicName + URI_SEPARATOR + DEFAULT_FEED_ID + URI_SEPARATOR + fileName;
 157         return new DefaultUriBuilderFactory().builder().scheme(dmaapProtocol).host(dmaapHostName).port(dmaapPortNumber)
 158                 .path(path).build();
 159     }
 160
 161     void setFileSystemResource(IFileSystemResource fileSystemResource) {
 162         fileResource = fileSystemResource;
 163     }
 164
 165     protected CloseableHttpAsyncClient getWebClient()
 166             throws KeyManagementException, NoSuchAlgorithmException, KeyStoreException {
 167         if (webClient != null) {
 168             return webClient;
 169         }
 170         SSLContext sslContext = null;
 171
 172         sslContext = new SSLContextBuilder().loadTrustMaterial(null, (certificate, authType) -> true).build();
 173
 174         //@formatter:off
 175         return HttpAsyncClients.custom()
 176                 .setSSLContext(sslContext)
 177                 .setSSLHostnameVerifier(new NoopHostnameVerifier())
 178                 .setRedirectStrategy(PublishRedirectStrategy.INSTANCE)
 179                 .build();
 180         //@formatter:on
 181     }
 182
 183     protected void setWebClient(CloseableHttpAsyncClient client) {
 184         this.webClient = client;
 185     }
 186 }