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