Added reactive DMaaPClient

Extracted WebCLientBuilder for
Producer and Consumer.
Added unit test for ReactiveProducerClient.

Change-Id: I632e6928813ed9feb48982900c173f741e4483e3
Issue-ID: DCAEGEN2-563
Signed-off-by: wasala <przemyslaw.wasala@nokia.com>

5 files changed, 183 insertions, 242 deletions

diff --git a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/DMaaPReactiveWebClient.java b/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/DMaaPReactiveWebClient.java
new file mode 100644
index 00000000..a41ec3a4
--- /dev/null
+++ b/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/DMaaPReactiveWebClient.java
@@ -0,0 +1,94 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.services.prh.service;
+
+import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpHeaders;
+import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/4/18
+ */
+public class DMaaPReactiveWebClient {
+
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    private DMaaPReactiveWebClient() {
+    }
+
+    private WebClient create(WebClientBuilder webClientBuilder) {
+        return WebClient.builder()
+            .defaultHeader(HttpHeaders.CONTENT_TYPE, webClientBuilder.dMaaPContentType)
+            .filter(basicAuthentication(webClientBuilder.dMaaPUserName, webClientBuilder.dMaaPUserPassword))
+            .filter(logRequest())
+            .filter(logResponse())
+            .build();
+    }
+
+    ExchangeFilterFunction logResponse() {
+        return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
+            logger.info("Response Status {}", clientResponse.statusCode());
+            return Mono.just(clientResponse);
+        });
+    }
+
+    ExchangeFilterFunction logRequest() {
+        return ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
+            logger.info("Request: {} {}", clientRequest.method(), clientRequest.url());
+            clientRequest.headers()
+                .forEach((name, values) -> values.forEach(value -> logger.info("{}={}", name, value)));
+            return Mono.just(clientRequest);
+        });
+    }
+
+    public static class WebClientBuilder {
+
+        private String dMaaPContentType;
+        private String dMaaPUserName;
+        private String dMaaPUserPassword;
+
+        public WebClientBuilder() {
+        }
+
+        public WebClientBuilder dmaapContentType(String dmaapContentType) {
+            this.dMaaPContentType = dmaapContentType;
+            return this;
+        }
+
+        public WebClientBuilder dmaapUserName(String dmaapUserName) {
+            this.dMaaPUserName = dmaapUserName;
+            return this;
+        }
+
+        public WebClientBuilder dmaapUserPassword(String dmaapUserPassword) {
+            this.dMaaPUserPassword = dmaapUserPassword;
+            return this;
+        }
+
+        public WebClient build() {
+            return new DMaaPReactiveWebClient().create(this);
+        }
+    }
+}
diff --git a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapHttpClientImpl.java b/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapHttpClientImpl.java
deleted file mode 100644
index 309106c5..00000000
--- a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapHttpClientImpl.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * PNF-REGISTRATION-HANDLER
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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.services.prh.service;
-
-import org.apache.http.auth.AuthScope;
-import org.apache.http.auth.Credentials;
-import org.apache.http.auth.UsernamePasswordCredentials;
-import org.apache.http.client.CredentialsProvider;
-import org.apache.http.impl.client.BasicCredentialsProvider;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.HttpClientBuilder;
-import org.apache.http.impl.client.HttpClients;
-import org.onap.dcaegen2.services.prh.config.DmaapCustomConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class DmaapHttpClientImpl {
-
-    private final Logger logger = LoggerFactory.getLogger(this.getClass());
-
-    private final String dmaapHostName;
-    private final Integer dmaapPortNumber;
-    private final String dmaapUserName;
-    private final String dmaapUserPassword;
-
-
-    public DmaapHttpClientImpl(DmaapCustomConfig configuration) {
-        this.dmaapHostName = configuration.dmaapHostName();
-        this.dmaapPortNumber = configuration.dmaapPortNumber();
-        this.dmaapUserName = configuration.dmaapUserName();
-        this.dmaapUserPassword = configuration.dmaapUserPassword();
-    }
-
-    public CloseableHttpClient getHttpClient() {
-
-        logger.info("Preparing closeable http client");
-
-        HttpClientBuilder httpClientBuilder = HttpClients.custom().useSystemProperties();
-
-        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
-
-        if (dmaapUserName != null) {
-            final AuthScope dmaapHostPortAuthScope = new AuthScope(dmaapHostName, dmaapPortNumber);
-            final Credentials dmaapCredentials = new UsernamePasswordCredentials(dmaapUserName, dmaapUserPassword);
-            credentialsProvider.setCredentials(dmaapHostPortAuthScope, dmaapCredentials);
-        }
-
-        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
-
-        return httpClientBuilder.build();
-    }
-}
diff --git a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/consumer/DmaapConsumerReactiveHttpClient.java b/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/consumer/DMaaPConsumerReactiveHttpClient.java
index cb7d5af2..23cdc98d 100644
--- a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/consumer/DmaapConsumerReactiveHttpClient.java
+++ b/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/consumer/DMaaPConsumerReactiveHttpClient.java
@@ -19,24 +19,20 @@
  */
 package org.onap.dcaegen2.services.prh.service.consumer;
 
-import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
-
 import java.net.URI;
 import java.net.URISyntaxException;
 import org.apache.http.client.utils.URIBuilder;
 import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
-import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
 import org.springframework.web.reactive.function.client.WebClient;
 import reactor.core.publisher.Mono;
 
 /**
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 6/26/18
  */
-public class DmaapConsumerReactiveHttpClient {
+public class DMaaPConsumerReactiveHttpClient {
 
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
@@ -47,32 +43,17 @@ public class DmaapConsumerReactiveHttpClient {
     private final String dmaapTopicName;
     private final String consumerGroup;
     private final String consumerId;
-    private final String dmaapContentType;
-    private final String dmaapUserName;
-    private final String dmaapUserPassword;
 
-    public DmaapConsumerReactiveHttpClient(DmaapConsumerConfiguration consumerConfiguration) {
+    public DMaaPConsumerReactiveHttpClient(DmaapConsumerConfiguration consumerConfiguration) {
         this.dmaapHostName = consumerConfiguration.dmaapHostName();
         this.dmaapProtocol = consumerConfiguration.dmaapProtocol();
         this.dmaapPortNumber = consumerConfiguration.dmaapPortNumber();
         this.dmaapTopicName = consumerConfiguration.dmaapTopicName();
         this.consumerGroup = consumerConfiguration.consumerGroup();
         this.consumerId = consumerConfiguration.consumerId();
-        this.dmaapContentType = consumerConfiguration.dmaapContentType();
-        this.dmaapUserName = consumerConfiguration.dmaapUserName();
-        this.dmaapUserPassword = consumerConfiguration.dmaapUserPassword();
-    }
-
-    public void initWebClient() {
-        this.webClient = WebClient.builder()
-            .defaultHeader(HttpHeaders.CONTENT_TYPE, dmaapContentType)
-            .filter(basicAuthentication(dmaapUserName, dmaapUserPassword))
-            .filter(logRequest())
-            .filter(logResponse())
-            .build();
     }
 
-    public Mono<String> getDmaaPConsumerResponse() {
+    public Mono<String> getDMaaPConsumerResponse() {
         try {
             return webClient
                 .get()
@@ -85,7 +66,7 @@ public class DmaapConsumerReactiveHttpClient {
                     Mono.error(new Exception("HTTP 500")))
                 .bodyToMono(String.class);
         } catch (URISyntaxException e) {
-            logger.warn("Exception while executing HTTP request: ", e);
+            logger.warn("Exception while evaluating URI ");
             return Mono.error(e);
         }
     }
@@ -94,28 +75,13 @@ public class DmaapConsumerReactiveHttpClient {
         return dmaapTopicName + "/" + consumerGroup + "/" + consumerId;
     }
 
-    void initWebClient(WebClient webClient) {
+    public DMaaPConsumerReactiveHttpClient createDMaaPWebClient(WebClient webClient) {
         this.webClient = webClient;
-    }
-
-    ExchangeFilterFunction logResponse() {
-        return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
-            logger.info("Response Status {}", clientResponse.statusCode());
-            return Mono.just(clientResponse);
-        });
+        return this;
     }
 
     URI getUri() throws URISyntaxException {
         return new URIBuilder().setScheme(dmaapProtocol).setHost(dmaapHostName).setPort(dmaapPortNumber)
             .setPath(createRequestPath()).build();
     }
-
-    ExchangeFilterFunction logRequest() {
-        return ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
-            logger.info("Request: {} {}", clientRequest.method(), clientRequest.url());
-            clientRequest.headers()
-                .forEach((name, values) -> values.forEach(value -> logger.info("{}={}", name, value)));
-            return Mono.just(clientRequest);
-        });
-    }
 }
diff --git a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/producer/DMaaPProducerReactiveHttpClient.java b/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/producer/DMaaPProducerReactiveHttpClient.java
new file mode 100644
index 00000000..e29ecc64
--- /dev/null
+++ b/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/producer/DMaaPProducerReactiveHttpClient.java
@@ -0,0 +1,83 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.services.prh.service.producer;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import org.apache.http.client.utils.URIBuilder;
+import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
+import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/4/18
+ */
+public class DMaaPProducerReactiveHttpClient {
+
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    private WebClient webClient;
+    private final String dmaapHostName;
+    private final Integer dmaapPortNumber;
+    private final String dmaapProtocol;
+    private final String dmaapTopicName;
+
+    public DMaaPProducerReactiveHttpClient(DmaapPublisherConfiguration dmaapPublisherConfiguration) {
+        this.dmaapHostName = dmaapPublisherConfiguration.dmaapHostName();
+        this.dmaapProtocol = dmaapPublisherConfiguration.dmaapProtocol();
+        this.dmaapPortNumber = dmaapPublisherConfiguration.dmaapPortNumber();
+        this.dmaapTopicName = dmaapPublisherConfiguration.dmaapTopicName();
+    }
+
+    public Mono<String> getDMaaPProducerResponse(Mono<ConsumerDmaapModel> consumerDmaapModelMono) {
+        try {
+            return webClient
+                .post()
+                .uri(getUri())
+                .body(BodyInserters.fromObject(consumerDmaapModelMono))
+                .retrieve()
+                .onStatus(HttpStatus::is4xxClientError, clientResponse ->
+                    Mono.error(new Exception("HTTP 400"))
+                )
+                .onStatus(HttpStatus::is5xxServerError, clientResponse ->
+                    Mono.error(new Exception("HTTP 500")))
+                .bodyToMono(String.class);
+        } catch (URISyntaxException e) {
+            logger.warn("Exception while evaluating URI");
+            return Mono.error(e);
+        }
+    }
+
+    public DMaaPProducerReactiveHttpClient createDMaaPWebClient(WebClient webClient) {
+        this.webClient = webClient;
+        return this;
+    }
+
+    URI getUri() throws URISyntaxException {
+        return new URIBuilder().setScheme(dmaapProtocol).setHost(dmaapHostName).setPort(dmaapPortNumber)
+            .setPath(dmaapTopicName).build();
+    }
+
+}
diff --git a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/producer/ExtendedDmaapProducerHttpClientImpl.java b/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/producer/ExtendedDmaapProducerHttpClientImpl.java
deleted file mode 100644
index 7bdaab1f..00000000
--- a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/producer/ExtendedDmaapProducerHttpClientImpl.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * PNF-REGISTRATION-HANDLER
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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.services.prh.service.producer;
-
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.client.methods.HttpRequestBase;
-import org.apache.http.client.utils.URIBuilder;
-import org.apache.http.entity.StringEntity;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.util.EntityUtils;
-import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
-import org.onap.dcaegen2.services.prh.model.CommonFunctions;
-import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.model.utils.HttpUtils;
-import org.onap.dcaegen2.services.prh.service.DmaapHttpClientImpl;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.Optional;
-
-public class ExtendedDmaapProducerHttpClientImpl {
-
-    private final Logger logger = LoggerFactory.getLogger(this.getClass());
-
-    private final CloseableHttpClient closeableHttpClient;
-    private final String dmaapHostName;
-    private final String dmaapProtocol;
-    private final Integer dmaapPortNumber;
-    private final String dmaapTopicName;
-    private final String dmaapContentType;
-    private ConsumerDmaapModel consumerDmaapModel;
-
-
-    public ExtendedDmaapProducerHttpClientImpl(DmaapPublisherConfiguration configuration) {
-        this.closeableHttpClient = new DmaapHttpClientImpl(configuration).getHttpClient();
-        this.dmaapHostName = configuration.dmaapHostName();
-        this.dmaapProtocol = configuration.dmaapProtocol();
-        this.dmaapPortNumber = configuration.dmaapPortNumber();
-        this.dmaapTopicName = configuration.dmaapTopicName();
-        this.dmaapContentType = configuration.dmaapContentType();
-    }
-
-    public Optional<Integer> getHttpProducerResponse(ConsumerDmaapModel consumerDmaapModel) {
-        this.consumerDmaapModel = consumerDmaapModel;
-        try {
-            return createRequest()
-                .flatMap(this::executeHttpClient);
-        } catch (URISyntaxException e) {
-            logger.warn("Exception while executing HTTP request: ", e);
-        }
-        return Optional.empty();
-    }
-
-    private Optional<Integer> executeHttpClient(HttpRequestBase httpRequestBase) {
-        try {
-            return closeableHttpClient.execute(httpRequestBase, this::handleResponse);
-        } catch (IOException e) {
-            logger.warn("Exception while executing HTTP request: ", e);
-        }
-        return Optional.empty();
-    }
-
-    private Optional<HttpRequestBase> createRequest() throws URISyntaxException {
-        return "application/json".equals(dmaapContentType)
-            ? createDmaapPublisherExtendedURI().map(this::createHttpPostRequest)
-            : Optional.empty();
-    }
-
-    private Optional<URI> createDmaapPublisherExtendedURI() throws URISyntaxException {
-        return Optional.ofNullable(new URIBuilder()
-            .setScheme(dmaapProtocol)
-            .setHost(dmaapHostName)
-            .setPort(dmaapPortNumber)
-            .setPath(dmaapTopicName).build());
-    }
-
-    private HttpPost createHttpPostRequest(URI extendedURI) {
-        HttpPost post = new HttpPost(extendedURI);
-        post.addHeader("Content-type", dmaapContentType);
-        createStringEntity().ifPresent(post::setEntity);
-        return post;
-    }
-
-    private Optional<StringEntity> createStringEntity() {
-        try {
-            return Optional.of(new StringEntity(CommonFunctions.createJsonBody(consumerDmaapModel)));
-        } catch (UnsupportedEncodingException | IllegalArgumentException e) {
-            logger.warn("Exception while parsing JSON: ", e);
-        }
-        return Optional.empty();
-    }
-
-    Optional<Integer> handleResponse(HttpResponse response) throws IOException {
-
-        final Integer responseCode = response.getStatusLine().getStatusCode();
-        logger.info("Status code of operation: {}", responseCode);
-        final HttpEntity responseEntity = response.getEntity();
-
-        if (HttpUtils.isSuccessfulResponseCode(responseCode)) {
-            logger.trace("HTTP response successful.");
-            return Optional.of(responseCode);
-        } else {
-            String aaiResponse = responseEntity != null ? EntityUtils.toString(responseEntity) : "";
-            logger.warn("HTTP response not successful : {}", aaiResponse);
-            return Optional.of(responseCode);
-        }
-    }
-}
\ No newline at end of file