diff options
Diffstat (limited to 'datafile-dmaap-client')
10 files changed, 416 insertions, 99 deletions
diff --git a/datafile-dmaap-client/pom.xml b/datafile-dmaap-client/pom.xml index 96e0988c..9f60f8da 100644 --- a/datafile-dmaap-client/pom.xml +++ b/datafile-dmaap-client/pom.xml @@ -68,6 +68,10 @@ <artifactId>spring-boot-starter-reactor-netty</artifactId> </dependency> <dependency> + <groupId>commons-io</groupId> + <artifactId>commons-io</artifactId> + </dependency> + <dependency> <groupId>org.junit.jupiter</groupId> <artifactId>junit-jupiter-api</artifactId> <scope>test</scope> diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/DmaapReactiveWebClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/DmaapReactiveWebClient.java index d5878b0d..095ba8c9 100644 --- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/DmaapReactiveWebClient.java +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/DmaapReactiveWebClient.java @@ -18,8 +18,6 @@ package org.onap.dcaegen2.collectors.datafile.service; -import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication; - import org.onap.dcaegen2.collectors.datafile.config.DmaapCustomConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -37,8 +35,6 @@ public class DmaapReactiveWebClient { private final Logger logger = LoggerFactory.getLogger(this.getClass()); private String dmaaPContentType; - private String dmaaPUserName; - private String dmaaPUserPassword; /** * Creating DmaapReactiveWebClient passing to them basic DmaapConfig. @@ -47,8 +43,6 @@ public class DmaapReactiveWebClient { * @return DmaapReactiveWebClient */ public DmaapReactiveWebClient fromConfiguration(DmaapCustomConfig dmaapCustomConfig) { - this.dmaaPUserName = dmaapCustomConfig.dmaapUserName(); - this.dmaaPUserPassword = dmaapCustomConfig.dmaapUserPassword(); this.dmaaPContentType = dmaapCustomConfig.dmaapContentType(); return this; } @@ -61,7 +55,6 @@ public class DmaapReactiveWebClient { public WebClient build() { return WebClient.builder() .defaultHeader(HttpHeaders.CONTENT_TYPE, dmaaPContentType) - .filter(basicAuthentication(dmaaPUserName, dmaaPUserPassword)) .filter(logRequest()) .filter(logResponse()) .build(); diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java index 36050ff7..fae86a82 100644 --- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java @@ -19,26 +19,29 @@ package org.onap.dcaegen2.collectors.datafile.service.producer; import com.google.gson.JsonElement; import com.google.gson.JsonParser; -import java.io.File; +import java.io.IOException; +import java.io.InputStream; import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.security.KeyManagementException; +import java.security.KeyStoreException; +import java.security.NoSuchAlgorithmException; -import org.apache.http.HttpHeaders; +import org.apache.commons.codec.binary.Base64; +import org.apache.commons.io.IOUtils; import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration; import org.onap.dcaegen2.collectors.datafile.model.CommonFunctions; import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.core.io.FileSystemResource; -import org.springframework.http.HttpStatus; -import org.springframework.web.reactive.function.BodyInserters; -import org.springframework.web.reactive.function.client.ClientResponse; -import org.springframework.web.reactive.function.client.WebClient; -import org.springframework.web.reactive.function.client.WebClient.RequestBodyUriSpec; -import org.springframework.web.reactive.function.client.WebClient.ResponseSpec; +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; import org.springframework.web.util.DefaultUriBuilderFactory; import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; /** * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/4/18 @@ -49,16 +52,21 @@ public class DmaapProducerReactiveHttpClient { private static final String X_ATT_DR_META = "X-ATT-DR-META"; private static final String NAME_JSON_TAG = "name"; private static final String LOCATION_JSON_TAG = "location"; + private static final String URI_SEPARATOR = "/"; private static final String DEFAULT_FEED_ID = "1"; private final Logger logger = LoggerFactory.getLogger(this.getClass()); - private WebClient webClient; private final String dmaapHostName; private final Integer dmaapPortNumber; private final String dmaapTopicName; private final String dmaapProtocol; private final String dmaapContentType; + private final String user; + private final String pwd; + + private IFileSystemResource fileResource; + private IRestTemplate restTemplate; /** * Constructor DmaapProducerReactiveHttpClient. @@ -72,65 +80,86 @@ public class DmaapProducerReactiveHttpClient { this.dmaapTopicName = dmaapPublisherConfiguration.dmaapTopicName(); this.dmaapProtocol = dmaapPublisherConfiguration.dmaapProtocol(); this.dmaapContentType = dmaapPublisherConfiguration.dmaapContentType(); + this.user = dmaapPublisherConfiguration.dmaapUserName(); + this.pwd = dmaapPublisherConfiguration.dmaapUserPassword(); } /** - * Function for calling DMaaP HTTP producer - post request to DMaaP. + * Function for calling DMaaP HTTP producer - post request to DMaaP DataRouter. * - * @param consumerDmaapModel - object which will be sent to DMaaP + * @param consumerDmaapModel - object which will be sent to DMaaP DataRouter * @return status code of operation */ public Flux<String> getDmaapProducerResponse(ConsumerDmaapModel consumerDmaapModel) { logger.trace("Entering getDmaapProducerResponse with {}", consumerDmaapModel); + try { + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.parseMediaType(dmaapContentType)); + addMetaDataToHead(consumerDmaapModel, headers); - RequestBodyUriSpec post = webClient.post(); + addUserCredentialsToHead(headers); - prepareHead(consumerDmaapModel, post); + HttpEntity<byte[]> request = addFileToRequest(consumerDmaapModel, headers); - prepareBody(consumerDmaapModel, post); - ResponseSpec responseSpec = post.retrieve(); - responseSpec.onStatus(HttpStatus::is4xxClientError, clientResponse -> handlePostErrors(consumerDmaapModel, clientResponse)); - responseSpec.onStatus(HttpStatus::is5xxServerError, clientResponse -> handlePostErrors(consumerDmaapModel, clientResponse)); - Flux<String> response = responseSpec.bodyToFlux(String.class); + logger.trace("Starting to publish to DR"); + ResponseEntity<String> responseEntity = getRestTemplate().exchange(getUri(consumerDmaapModel.getName()), + HttpMethod.PUT, request, String.class); - logger.trace("Exiting getDmaapProducerResponse with {}", response); - return response; + return Flux.just(responseEntity.getStatusCode().toString()); + } catch (Exception e) { + logger.error("Unable to send file to DataRouter. Data: {}", consumerDmaapModel, e); + return Flux.empty(); + } } - public DmaapProducerReactiveHttpClient createDmaapWebClient(WebClient webClient) { - this.webClient = webClient; - return this; + private void addUserCredentialsToHead(HttpHeaders headers) { + String plainCreds = user + ":" + pwd; + byte[] plainCredsBytes = plainCreds.getBytes(StandardCharsets.ISO_8859_1); + byte[] base64CredsBytes = Base64.encodeBase64(plainCredsBytes); + String base64Creds = new String(base64CredsBytes); + logger.trace("base64Creds...: {}", base64Creds); + headers.add("Authorization", "Basic " + base64Creds); } - private void prepareHead(ConsumerDmaapModel model, RequestBodyUriSpec post) { - post.header(HttpHeaders.CONTENT_TYPE, dmaapContentType); - - JsonElement metaData = new JsonParser().parse(CommonFunctions.createJsonBody(model)); - String name = metaData.getAsJsonObject().remove(NAME_JSON_TAG).getAsString(); + private void addMetaDataToHead(ConsumerDmaapModel consumerDmaapModel, HttpHeaders headers) { + JsonElement metaData = new JsonParser().parse(CommonFunctions.createJsonBody(consumerDmaapModel)); + metaData.getAsJsonObject().remove(NAME_JSON_TAG).getAsString(); metaData.getAsJsonObject().remove(LOCATION_JSON_TAG); - post.header(X_ATT_DR_META, metaData.toString()); + headers.set(X_ATT_DR_META, metaData.toString()); + } - post.uri(getUri(name)); + private HttpEntity<byte[]> addFileToRequest(ConsumerDmaapModel consumerDmaapModel, HttpHeaders headers) + throws IOException { + InputStream in = getInputStream(consumerDmaapModel.getLocation()); + return new HttpEntity<>(IOUtils.toByteArray(in), headers); } - private void prepareBody(ConsumerDmaapModel model, RequestBodyUriSpec post) { - String fileLocation = model.getLocation(); - File fileResource = new File(fileLocation); - FileSystemResource httpResource = new FileSystemResource(fileResource); - post.body(BodyInserters.fromResource(httpResource)); + private InputStream getInputStream(String filePath) throws IOException { + if (fileResource == null) { + fileResource = new FileSystemResourceWrapper(filePath); + } + return fileResource.getInputStream(); + } + + private IRestTemplate getRestTemplate() throws NoSuchAlgorithmException, KeyManagementException, KeyStoreException { + if (restTemplate == null) { + restTemplate = new RestTemplateWrapper(); + } + return restTemplate; } private URI getUri(String fileName) { - String path = dmaapTopicName + "/" + DEFAULT_FEED_ID + "/" + fileName; + String path = dmaapTopicName + URI_SEPARATOR + DEFAULT_FEED_ID + URI_SEPARATOR + fileName; return new DefaultUriBuilderFactory().builder().scheme(dmaapProtocol).host(dmaapHostName).port(dmaapPortNumber) .path(path).build(); } - private Mono<Exception> handlePostErrors(ConsumerDmaapModel model, ClientResponse clientResponse) { - String errorMessage = "Unable to post file to Data Router. " + model + "Reason: " + clientResponse.toString(); - logger.error(errorMessage); + protected void setFileSystemResource(IFileSystemResource fileSystemResource) { + fileResource = fileSystemResource; + } - return Mono.error(new Exception(errorMessage)); + protected void setRestTemplate(IRestTemplate restTemplate) { + this.restTemplate = restTemplate; } } diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/FileSystemResourceWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/FileSystemResourceWrapper.java new file mode 100644 index 00000000..63496916 --- /dev/null +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/FileSystemResourceWrapper.java @@ -0,0 +1,40 @@ +/* + * ============LICENSE_START====================================================================== + * Copyright (C) 2018 Nordix Foundation. All rights reserved. + * =============================================================================================== + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END======================================================================== + */ + +package org.onap.dcaegen2.collectors.datafile.service.producer; + +import java.io.IOException; +import java.io.InputStream; + +import org.springframework.core.io.FileSystemResource; + +/** + * @author + * + */ +public class FileSystemResourceWrapper implements IFileSystemResource { + private FileSystemResource realResource; + + public FileSystemResourceWrapper(String path) { + realResource = new FileSystemResource(path); + } + @Override + public InputStream getInputStream() throws IOException { + return realResource.getInputStream(); + } +} diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/IFileSystemResource.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/IFileSystemResource.java new file mode 100644 index 00000000..6ecb1c6e --- /dev/null +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/IFileSystemResource.java @@ -0,0 +1,29 @@ +/* + * ============LICENSE_START====================================================================== + * Copyright (C) 2018 Nordix Foundation. All rights reserved. + * =============================================================================================== + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License + * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the License for the specific language governing permissions and limitations under + * the License. + * ============LICENSE_END======================================================================== + */ + +package org.onap.dcaegen2.collectors.datafile.service.producer; + +import java.io.IOException; +import java.io.InputStream; + +/** + * @author + * + */ +public interface IFileSystemResource { + + InputStream getInputStream() throws IOException; +} diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/IRestTemplate.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/IRestTemplate.java new file mode 100644 index 00000000..7f8d3b56 --- /dev/null +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/IRestTemplate.java @@ -0,0 +1,32 @@ +/* + * ============LICENSE_START====================================================================== + * Copyright (C) 2018 Nordix Foundation. All rights reserved. + * =============================================================================================== + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License + * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the License for the specific language governing permissions and limitations under + * the License. + * ============LICENSE_END======================================================================== + */ + +package org.onap.dcaegen2.collectors.datafile.service.producer; + +import java.net.URI; + +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpMethod; +import org.springframework.http.ResponseEntity; + +/** + * @author + * + */ +public interface IRestTemplate { + public ResponseEntity<String> exchange(URI url, HttpMethod method, HttpEntity<byte[]> requestEntity, + Class<String> responseType); +} diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/PublishRedirectStrategy.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/PublishRedirectStrategy.java new file mode 100644 index 00000000..ff3a03bf --- /dev/null +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/PublishRedirectStrategy.java @@ -0,0 +1,81 @@ +/* + * ============LICENSE_START====================================================================== + * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved. + * =============================================================================================== + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License + * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the License for the specific language governing permissions and limitations under + * the License. + * ============LICENSE_END======================================================================== + */ +package org.onap.dcaegen2.collectors.datafile.service.producer; + +import java.net.URI; + +import org.apache.http.HttpRequest; +import org.apache.http.HttpResponse; +import org.apache.http.ProtocolException; +import org.apache.http.annotation.Contract; +import org.apache.http.annotation.ThreadingBehavior; +import org.apache.http.client.methods.HttpDelete; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.methods.HttpHead; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.client.methods.HttpPut; +import org.apache.http.client.methods.HttpUriRequest; +import org.apache.http.client.methods.RequestBuilder; +import org.apache.http.impl.client.DefaultRedirectStrategy; +import org.apache.http.protocol.HttpContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * PublishRedirectStrategy implementation + * that automatically redirects all HEAD, GET, POST, PUT, and DELETE requests. + * This strategy relaxes restrictions on automatic redirection of + * POST methods imposed by the HTTP specification. + * + */ +@Contract(threading = ThreadingBehavior.IMMUTABLE) +public class PublishRedirectStrategy extends DefaultRedirectStrategy { + + public static final PublishRedirectStrategy INSTANCE = new PublishRedirectStrategy(); + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + /** + * Redirectable methods. + */ + private static final String[] REDIRECT_METHODS = new String[] { + HttpPut.METHOD_NAME, + HttpGet.METHOD_NAME, + HttpPost.METHOD_NAME, + HttpHead.METHOD_NAME, + HttpDelete.METHOD_NAME + }; + + @Override + protected boolean isRedirectable(final String method) { + for (final String m: REDIRECT_METHODS) { + if (m.equalsIgnoreCase(method)) { + return true; + } + } + return false; + } + + @Override + public HttpUriRequest getRedirect( + final HttpRequest request, + final HttpResponse response, + final HttpContext context) throws ProtocolException { + final URI uri = getLocationURI(request, response, context); + logger.trace("getRedirect...: {}", request.toString()); + return RequestBuilder.copy(request).setUri(uri).build(); + } + +} diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/RequestResponseLoggingInterceptor.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/RequestResponseLoggingInterceptor.java new file mode 100644 index 00000000..bf40700e --- /dev/null +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/RequestResponseLoggingInterceptor.java @@ -0,0 +1,61 @@ +/* + * ============LICENSE_START====================================================================== + * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved. + * =============================================================================================== + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License + * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the License for the specific language governing permissions and limitations under + * the License. + * ============LICENSE_END======================================================================== + */ +package org.onap.dcaegen2.collectors.datafile.service.producer; +import java.io.IOException; +import java.nio.charset.Charset; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.http.HttpRequest; +import org.springframework.http.client.ClientHttpRequestExecution; +import org.springframework.http.client.ClientHttpRequestInterceptor; +import org.springframework.http.client.ClientHttpResponse; +import org.springframework.util.StreamUtils; + +public class RequestResponseLoggingInterceptor implements ClientHttpRequestInterceptor { + + private final Logger log = LoggerFactory.getLogger(this.getClass()); + + @Override + public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution) throws IOException { + logRequest(request, body); + ClientHttpResponse response = execution.execute(request, body); + logResponse(response); + return response; + } + + private void logRequest(HttpRequest request, byte[] body) throws IOException { + if (log.isDebugEnabled()) { + log.debug("===========================request begin================================================"); + log.debug("URI : {}", request.getURI()); + log.debug("Method : {}", request.getMethod()); + log.debug("Headers : {}", request.getHeaders()); + log.debug("Request body: {}", new String(body, "UTF-8")); + log.debug("==========================request end================================================"); + } + } + + private void logResponse(ClientHttpResponse response) throws IOException { + if (log.isDebugEnabled()) { + log.debug("============================response begin=========================================="); + log.debug("Status code : {}", response.getStatusCode()); + log.debug("Status text : {}", response.getStatusText()); + log.debug("Headers : {}", response.getHeaders()); + log.debug("Response body: {}", StreamUtils.copyToString(response.getBody(), Charset.defaultCharset())); + log.debug("=======================response end================================================="); + } + } +}
\ No newline at end of file diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/RestTemplateWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/RestTemplateWrapper.java new file mode 100644 index 00000000..298f6693 --- /dev/null +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/RestTemplateWrapper.java @@ -0,0 +1,65 @@ +/* + * ============LICENSE_START====================================================================== + * Copyright (C) 2018 Nordix Foundation. All rights reserved. + * =============================================================================================== + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License + * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the License for the specific language governing permissions and limitations under + * the License. + * ============LICENSE_END======================================================================== + */ + +package org.onap.dcaegen2.collectors.datafile.service.producer; + +import java.net.URI; +import java.security.KeyManagementException; +import java.security.KeyStoreException; +import java.security.NoSuchAlgorithmException; +import java.util.Collections; + +import javax.net.ssl.SSLContext; + +import org.apache.http.conn.ssl.NoopHostnameVerifier; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.ssl.SSLContextBuilder; +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpMethod; +import org.springframework.http.ResponseEntity; +import org.springframework.http.client.HttpComponentsClientHttpRequestFactory; +import org.springframework.web.client.RestTemplate; + +/** + * @author + * + */ +public class RestTemplateWrapper implements IRestTemplate { + private RestTemplate restTemplate; + + public RestTemplateWrapper() throws KeyManagementException, NoSuchAlgorithmException, KeyStoreException { + SSLContext sslContext = + new SSLContextBuilder().loadTrustMaterial(null, (certificate, authType) -> true).build(); + CloseableHttpClient httpClient = + HttpClients.custom().setSSLContext(sslContext).setSSLHostnameVerifier(new NoopHostnameVerifier()) + .setRedirectStrategy(new PublishRedirectStrategy()).build(); + + HttpComponentsClientHttpRequestFactory requestFactory = new HttpComponentsClientHttpRequestFactory(); + requestFactory.setHttpClient(httpClient); + + restTemplate = new RestTemplate(requestFactory); + restTemplate.setInterceptors(Collections.singletonList(new RequestResponseLoggingInterceptor())); + + } + + @Override + public ResponseEntity<String> exchange(URI url, HttpMethod method, HttpEntity<byte[]> requestEntity, + Class<String> responseType) { + return restTemplate.exchange(url, method, requestEntity, responseType); + } + +} diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java index 5f4c1a58..5dbc9089 100644 --- a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java +++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java @@ -17,32 +17,26 @@ package org.onap.dcaegen2.collectors.datafile.service.producer; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; -import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication; - -import com.google.gson.JsonElement; -import com.google.gson.JsonParser; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; import java.net.URI; -import java.util.ArrayList; -import java.util.List; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration; -import org.onap.dcaegen2.collectors.datafile.model.CommonFunctions; import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel; import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModelForUnitTest; -import org.springframework.http.HttpHeaders; -import org.springframework.web.reactive.function.client.WebClient; -import org.springframework.web.reactive.function.client.WebClient.RequestBodyUriSpec; -import org.springframework.web.reactive.function.client.WebClient.ResponseSpec; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; import org.springframework.web.util.DefaultUriBuilderFactory; -import reactor.core.publisher.Flux; import reactor.test.StepVerifier; /** @@ -57,70 +51,59 @@ class DmaapProducerReactiveHttpClientTest { private static final String X_ATT_DR_META = "X-ATT-DR-META"; private static final String HOST = "54.45.33.2"; - private static final String HTTP_SCHEME = "http"; + private static final String HTTPS_SCHEME = "https"; private static final int PORT = 1234; private static final String APPLICATION_OCTET_STREAM_CONTENT_TYPE = "application/octet-stream"; + private static final String URI_SEPARATOR = "/"; private static final String PUBLISH_TOPIC = "publish"; private static final String DEFAULT_FEED_ID = "1"; + private static final String FILE_CONTENT = "Just a string."; private DmaapProducerReactiveHttpClient dmaapProducerReactiveHttpClient; private DmaapPublisherConfiguration dmaapPublisherConfigurationMock = mock(DmaapPublisherConfiguration.class); private ConsumerDmaapModel consumerDmaapModel = new ConsumerDmaapModelForUnitTest(); - private WebClient webClientMock = mock(WebClient.class); - private RequestBodyUriSpec requestBodyUriSpecMock; - private ResponseSpec responseSpecMock; + + private IFileSystemResource fileSystemResourceMock = mock(IFileSystemResource.class); + private IRestTemplate restTemplateMock = mock(IRestTemplate.class); + private ResponseEntity<String> responseEntityMock = mock(ResponseEntity.class); @BeforeEach void setUp() { when(dmaapPublisherConfigurationMock.dmaapHostName()).thenReturn(HOST); - when(dmaapPublisherConfigurationMock.dmaapProtocol()).thenReturn(HTTP_SCHEME); + when(dmaapPublisherConfigurationMock.dmaapProtocol()).thenReturn(HTTPS_SCHEME); when(dmaapPublisherConfigurationMock.dmaapPortNumber()).thenReturn(PORT); - when(dmaapPublisherConfigurationMock.dmaapUserName()).thenReturn("DATAFILE"); - when(dmaapPublisherConfigurationMock.dmaapUserPassword()).thenReturn("DATAFILE"); + when(dmaapPublisherConfigurationMock.dmaapUserName()).thenReturn("dradmin"); + when(dmaapPublisherConfigurationMock.dmaapUserPassword()).thenReturn("dradmin"); when(dmaapPublisherConfigurationMock.dmaapContentType()).thenReturn(APPLICATION_OCTET_STREAM_CONTENT_TYPE); when(dmaapPublisherConfigurationMock.dmaapTopicName()).thenReturn(PUBLISH_TOPIC); dmaapProducerReactiveHttpClient = new DmaapProducerReactiveHttpClient(dmaapPublisherConfigurationMock); - - webClientMock = spy(WebClient.builder() - .defaultHeader(HttpHeaders.CONTENT_TYPE, dmaapPublisherConfigurationMock.dmaapContentType()) - .filter(basicAuthentication(dmaapPublisherConfigurationMock.dmaapUserName(), - dmaapPublisherConfigurationMock.dmaapUserPassword())) - .build()); - requestBodyUriSpecMock = mock(RequestBodyUriSpec.class); - responseSpecMock = mock(ResponseSpec.class); + dmaapProducerReactiveHttpClient.setFileSystemResource(fileSystemResourceMock); + dmaapProducerReactiveHttpClient.setRestTemplate(restTemplateMock); } @Test - void getHttpResponse_Success() { + void getHttpResponse_Success() throws Exception { mockWebClientDependantObject(); - dmaapProducerReactiveHttpClient.createDmaapWebClient(webClientMock); - List<ConsumerDmaapModel> consumerDmaapModelList = new ArrayList<ConsumerDmaapModel>(); - consumerDmaapModelList.add(consumerDmaapModel); StepVerifier.create(dmaapProducerReactiveHttpClient.getDmaapProducerResponse(consumerDmaapModel)) - .expectNext("200").verifyComplete(); - - verify(requestBodyUriSpecMock).header(HttpHeaders.CONTENT_TYPE, APPLICATION_OCTET_STREAM_CONTENT_TYPE); - JsonElement metaData = new JsonParser().parse(CommonFunctions.createJsonBody(consumerDmaapModel)); - metaData.getAsJsonObject().remove(LOCATION_JSON_TAG); - metaData.getAsJsonObject().remove(NAME_JSON_TAG); - verify(requestBodyUriSpecMock).header(X_ATT_DR_META, metaData.toString()); - URI expectedUri = new DefaultUriBuilderFactory().builder().scheme(HTTP_SCHEME).host(HOST).port(PORT) - .path(PUBLISH_TOPIC + "/" + DEFAULT_FEED_ID + "/" + FILE_NAME).build(); - verify(requestBodyUriSpecMock).uri(expectedUri); - verify(requestBodyUriSpecMock).body(any()); + .expectNext(HttpStatus.OK.toString()).verifyComplete(); + + URI expectedUri = new DefaultUriBuilderFactory().builder().scheme(HTTPS_SCHEME).host(HOST).port(PORT) + .path(PUBLISH_TOPIC + URI_SEPARATOR + DEFAULT_FEED_ID + URI_SEPARATOR + FILE_NAME).build(); + + verify(restTemplateMock) + .exchange(eq(expectedUri), eq(HttpMethod.PUT), any(), eq(String.class)); } - private void mockWebClientDependantObject() { - when(webClientMock.post()).thenReturn(requestBodyUriSpecMock); - when(requestBodyUriSpecMock.uri((URI) any())).thenReturn(requestBodyUriSpecMock); + private void mockWebClientDependantObject() throws IOException { + InputStream fileStream = new ByteArrayInputStream(FILE_CONTENT.getBytes()); + when(fileSystemResourceMock.getInputStream()).thenReturn(fileStream); - when(requestBodyUriSpecMock.retrieve()).thenReturn(responseSpecMock); - when(responseSpecMock.onStatus(any(), any())).thenReturn(responseSpecMock); - Flux<String> expectedResult = Flux.just("200"); - when(responseSpecMock.bodyToFlux(String.class)).thenReturn(expectedResult); + when(restTemplateMock.exchange(any(), any(), any(), any())) + .thenReturn(responseEntityMock); + when(responseEntityMock.getStatusCode()).thenReturn(HttpStatus.OK); } } |