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