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.security.KeyManagementException;
  27 import java.security.KeyStoreException;
  28 import java.security.NoSuchAlgorithmException;
  29
  30 import org.apache.commons.codec.binary.Base64;
  31 import org.apache.commons.io.IOUtils;
  32 import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
  33 import org.onap.dcaegen2.collectors.datafile.model.CommonFunctions;
  34 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
  35 import org.slf4j.Logger;
  36 import org.slf4j.LoggerFactory;
  37 import org.springframework.http.HttpEntity;
  38 import org.springframework.http.HttpHeaders;
  39 import org.springframework.http.HttpMethod;
  40 import org.springframework.http.MediaType;
  41 import org.springframework.http.ResponseEntity;
  42 import org.springframework.web.util.DefaultUriBuilderFactory;
  43
  44 import reactor.core.publisher.Flux;
  45
  46 /**
  47  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/4/18
  48  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  49  */
  50 public class DmaapProducerReactiveHttpClient {
  51
  52     private static final String X_ATT_DR_META = "X-ATT-DR-META";
  53     private static final String NAME_JSON_TAG = "name";
  54     private static final String LOCATION_JSON_TAG = "location";
  55     private static final String URI_SEPARATOR = "/";
  56     private static final String DEFAULT_FEED_ID = "1";
  57
  58     private final Logger logger = LoggerFactory.getLogger(this.getClass());
  59
  60     private final String dmaapHostName;
  61     private final Integer dmaapPortNumber;
  62     private final String dmaapTopicName;
  63     private final String dmaapProtocol;
  64     private final String dmaapContentType;
  65     private final String user;
  66     private final String pwd;
  67
  68     private IFileSystemResource fileResource;
  69     private IRestTemplate restTemplate;
  70
  71     /**
  72      * Constructor DmaapProducerReactiveHttpClient.
  73      *
  74      * @param dmaapPublisherConfiguration - DMaaP producer configuration object
  75      */
  76     public DmaapProducerReactiveHttpClient(DmaapPublisherConfiguration dmaapPublisherConfiguration) {
  77
  78         this.dmaapHostName = dmaapPublisherConfiguration.dmaapHostName();
  79         this.dmaapPortNumber = dmaapPublisherConfiguration.dmaapPortNumber();
  80         this.dmaapTopicName = dmaapPublisherConfiguration.dmaapTopicName();
  81         this.dmaapProtocol = dmaapPublisherConfiguration.dmaapProtocol();
  82         this.dmaapContentType = dmaapPublisherConfiguration.dmaapContentType();
  83         this.user = dmaapPublisherConfiguration.dmaapUserName();
  84         this.pwd = dmaapPublisherConfiguration.dmaapUserPassword();
  85     }
  86
  87     /**
  88      * Function for calling DMaaP HTTP producer - post request to DMaaP DataRouter.
  89      *
  90      * @param consumerDmaapModel - object which will be sent to DMaaP DataRouter
  91      * @return status code of operation
  92      */
  93     public Flux<String> getDmaapProducerResponse(ConsumerDmaapModel consumerDmaapModel) {
  94         logger.trace("Entering getDmaapProducerResponse with {}", consumerDmaapModel);
  95         try {
  96             HttpHeaders headers = new HttpHeaders();
  97             headers.setContentType(MediaType.parseMediaType(dmaapContentType));
  98             addMetaDataToHead(consumerDmaapModel, headers);
  99
 100             addUserCredentialsToHead(headers);
 101
 102             InputStream fileInputStream = getInputStream(consumerDmaapModel.getLocation());
 103             HttpEntity<byte[]> request = addFileToRequest(fileInputStream, headers);
 104
 105
 106             logger.trace("Starting to publish to DR");
 107             ResponseEntity<String> responseEntity = getRestTemplate().exchange(getUri(consumerDmaapModel.getName()),
 108                     HttpMethod.PUT, request, String.class);
 109
 110             return Flux.just(responseEntity.getStatusCode().toString());
 111         } catch (Exception e) {
 112             logger.error("Unable to send file to DataRouter. Data: {}", consumerDmaapModel, e);
 113             return Flux.empty();
 114         }
 115     }
 116
 117     private void addUserCredentialsToHead(HttpHeaders headers) {
 118         String plainCreds = user + ":" + pwd;
 119         byte[] plainCredsBytes = plainCreds.getBytes(StandardCharsets.ISO_8859_1);
 120         byte[] base64CredsBytes = Base64.encodeBase64(plainCredsBytes);
 121         String base64Creds = new String(base64CredsBytes);
 122         logger.trace("base64Creds...: {}", base64Creds);
 123         headers.add("Authorization", "Basic " + base64Creds);
 124     }
 125
 126     private void addMetaDataToHead(ConsumerDmaapModel consumerDmaapModel, HttpHeaders headers) {
 127         JsonElement metaData = new JsonParser().parse(CommonFunctions.createJsonBody(consumerDmaapModel));
 128         metaData.getAsJsonObject().remove(NAME_JSON_TAG).getAsString();
 129         metaData.getAsJsonObject().remove(LOCATION_JSON_TAG);
 130         headers.set(X_ATT_DR_META, metaData.toString());
 131     }
 132
 133     private HttpEntity<byte[]> addFileToRequest(InputStream inputStream, HttpHeaders headers)
 134             throws IOException {
 135         return new HttpEntity<>(IOUtils.toByteArray(inputStream), headers);
 136     }
 137
 138     private InputStream getInputStream(String filePath) throws IOException {
 139         if (fileResource == null) {
 140             fileResource = new FileSystemResourceWrapper(filePath);
 141         }
 142         return fileResource.getInputStream();
 143     }
 144
 145     private IRestTemplate getRestTemplate() throws NoSuchAlgorithmException, KeyManagementException, KeyStoreException {
 146         if (restTemplate == null) {
 147             restTemplate = new RestTemplateWrapper();
 148         }
 149         return restTemplate;
 150     }
 151
 152     private URI getUri(String fileName) {
 153         String path = dmaapTopicName + URI_SEPARATOR + DEFAULT_FEED_ID + URI_SEPARATOR + fileName;
 154         return new DefaultUriBuilderFactory().builder().scheme(dmaapProtocol).host(dmaapHostName).port(dmaapPortNumber)
 155                 .path(path).build();
 156     }
 157
 158     protected void setFileSystemResource(IFileSystemResource fileSystemResource) {
 159         fileResource = fileSystemResource;
 160     }
 161
 162     protected void setRestTemplate(IRestTemplate restTemplate) {
 163         this.restTemplate = restTemplate;
 164     }
 165 }