Add seed code.

First version based on PRH micro service.

Change-Id: Iea1673a8a1961006b1ea98ef245e213e3652eb82
Issue-ID: DCAEGEN2-638
Signed-off-by: elinuxhenrik <henrik.b.andersson@est.tech>

12 files changed, 997 insertions, 0 deletions

diff --git a/datafile-aai-client/pom.xml b/datafile-aai-client/pom.xml
new file mode 100644
index 00000000..bc496562
--- /dev/null
+++ b/datafile-aai-client/pom.xml
@@ -0,0 +1,130 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ ============LICENSE_START=======================================================
+  ~ Datafile Collector Service
+  ~ ================================================================================
+  ~ 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=========================================================
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <artifactId>datafile</artifactId>
+    <groupId>org.onap.dcaegen2.collectors</groupId>
+    <version>1.0.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>datafile-aai-client</artifactId>
+  <groupId>org.onap.dcaegen2.collectors.datafile</groupId>
+  <packaging>jar</packaging>
+
+
+  <properties>
+    <main.basedir>${project.parent.basedir}</main.basedir>
+  </properties>
+  <dependencies>
+
+    <!-- DEVELOPMENT DEPENDENCIES -->
+    <dependency>
+      <groupId>org.immutables</groupId>
+      <artifactId>value</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.immutables</groupId>
+      <artifactId>gson</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework</groupId>
+      <artifactId>spring-webflux</artifactId>
+      <version>5.0.5.RELEASE</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.httpcomponents</groupId>
+      <artifactId>httpclient</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework</groupId>
+      <artifactId>spring-beans</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework</groupId>
+      <artifactId>spring-context</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-reactor-netty</artifactId>
+      <version>2.0.4.RELEASE</version>
+    </dependency>
+    <dependency>
+      <groupId>org.onap.dcaegen2.collectors.datafile</groupId>
+      <artifactId>datafile-commons</artifactId>
+      <version>1.0.0-SNAPSHOT</version>
+    </dependency>
+
+    <!-- LOGGING DEPENDENCIES-->
+    <dependency>
+      <groupId>ch.qos.logback</groupId>
+      <artifactId>logback-classic</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>jul-to-slf4j</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>log4j-over-slf4j</artifactId>
+    </dependency>
+
+    <!-- TEST DEPENDENCIES-->
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-api</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-engine</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.junit.vintage</groupId>
+      <artifactId>junit-vintage-engine</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.projectreactor</groupId>
+      <artifactId>reactor-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>
diff --git a/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/config/AaiClientConfiguration.java b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/config/AaiClientConfiguration.java
new file mode 100644
index 00000000..ce77599a
--- /dev/null
+++ b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/config/AaiClientConfiguration.java
@@ -0,0 +1,67 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Datafile Collector Service
+ * ================================================================================
+ * 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.collectors.datafile.config;
+
+import java.io.Serializable;
+
+import java.util.Map;
+
+import org.immutables.gson.Gson;
+import org.immutables.value.Value;
+import org.springframework.stereotype.Component;
+
+
+@Component
+@Value.Immutable(prehash = true)
+@Value.Style(builder = "new")
+@Gson.TypeAdapters
+public abstract class AaiClientConfiguration implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    @Value.Parameter
+    public abstract String aaiHost();
+
+    @Value.Parameter
+    public abstract Integer aaiPort();
+
+    @Value.Parameter
+    public abstract String aaiProtocol();
+
+    @Value.Parameter
+    public abstract String aaiUserName();
+
+    @Value.Parameter
+    public abstract String aaiUserPassword();
+
+    @Value.Parameter
+    public abstract Boolean aaiIgnoreSslCertificateErrors();
+
+    @Value.Parameter
+    public abstract String aaiBasePath();
+
+    @Value.Parameter
+    public abstract String aaiPnfPath();
+
+    @Value.Parameter
+    public abstract Map<String, String> aaiHeaders();
+
+}
diff --git a/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/AaiRequestException.java b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/AaiRequestException.java
new file mode 100644
index 00000000..a30867a2
--- /dev/null
+++ b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/AaiRequestException.java
@@ -0,0 +1,28 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Datafile Collector Service
+ * ================================================================================
+ * 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.collectors.datafile.exceptions;
+
+public class AaiRequestException extends Exception {
+
+    public AaiRequestException(String message) {
+        super(message);
+    }
+}
diff --git a/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiClient.java b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiClient.java
new file mode 100644
index 00000000..5ab9b4a6
--- /dev/null
+++ b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiClient.java
@@ -0,0 +1,29 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Datafile Collector Service
+ * ================================================================================
+ * 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.collectors.datafile.service;
+
+import org.apache.http.impl.client.CloseableHttpClient;
+
+@FunctionalInterface
+public interface AaiClient {
+    CloseableHttpClient getAaiHttpClient();
+}
+
diff --git a/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiClientImpl.java b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiClientImpl.java
new file mode 100644
index 00000000..8af4a13a
--- /dev/null
+++ b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiClientImpl.java
@@ -0,0 +1,94 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Datafile Collector Service
+ * ================================================================================
+ * 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.collectors.datafile.service;
+
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+
+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.conn.ssl.NoopHostnameVerifier;
+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.apache.http.ssl.SSLContextBuilder;
+import org.apache.http.ssl.TrustStrategy;
+import org.onap.dcaegen2.collectors.datafile.config.AaiClientConfiguration;
+import org.onap.dcaegen2.collectors.datafile.service.AaiClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class AaiClientImpl implements AaiClient {
+
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    private AaiClientConfiguration aaiClientConfig;
+
+
+    AaiClientImpl(AaiClientConfiguration aaiClientConfiguration) {
+        this.aaiClientConfig = aaiClientConfiguration;
+    }
+
+    @Override
+    public CloseableHttpClient getAaiHttpClient() {
+
+        final HttpClientBuilder httpClientBuilder = HttpClients.custom().useSystemProperties();
+        final boolean aaiIgnoreSslCertificateErrors = aaiClientConfig.aaiIgnoreSslCertificateErrors();
+
+        TrustStrategy acceptingTrustStrategy = (cert, authType) -> true;
+
+        if (aaiIgnoreSslCertificateErrors) {
+            try {
+                logger.info("Setting SSL Context for AAI HTTP Client");
+                httpClientBuilder.setSSLContext(new SSLContextBuilder()
+                    .loadTrustMaterial(null, acceptingTrustStrategy)
+                    .build());
+
+            } catch (NoSuchAlgorithmException | KeyStoreException | KeyManagementException e) {
+                logger.error("Exception while setting SSL Context for AAI HTTP Client: {}", e);
+            }
+
+            httpClientBuilder.setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE);
+        }
+
+        final String aaiUserName = aaiClientConfig.aaiUserName();
+
+        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+
+        if (aaiUserName != null) {
+            final String aaiHost = aaiClientConfig.aaiHost();
+            final Integer aaiHostPortNumber = aaiClientConfig.aaiPort();
+            final String aaiUserPassword = aaiClientConfig.aaiUserPassword();
+            final AuthScope aaiHostPortAuthScope = new AuthScope(aaiHost, aaiHostPortNumber);
+            final Credentials aaiCredentials = new UsernamePasswordCredentials(aaiUserName, aaiUserPassword);
+            credentialsProvider.setCredentials(aaiHostPortAuthScope, aaiCredentials);
+        }
+
+        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
+
+        return httpClientBuilder.build();
+    }
+}
diff --git a/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiConsumerClient.java b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiConsumerClient.java
new file mode 100644
index 00000000..ecc9ed2c
--- /dev/null
+++ b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiConsumerClient.java
@@ -0,0 +1,140 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Datafile Collector Service
+ * ================================================================================
+ * 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.collectors.datafile.service;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+import org.apache.http.HttpEntity;
+import org.apache.http.client.ResponseHandler;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpRequestBase;
+import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.util.EntityUtils;
+import org.onap.dcaegen2.collectors.datafile.config.AaiClientConfiguration;
+import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.model.utils.HttpUtils;
+import org.onap.dcaegen2.collectors.datafile.service.AaiClientImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class AaiConsumerClient {
+
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    private final CloseableHttpClient closeableHttpClient;
+    private final String aaiHost;
+    private final String aaiProtocol;
+    private final Integer aaiHostPortNumber;
+    private final String aaiPath;
+    private final Map<String, String> aaiHeaders;
+
+    /**
+     * A{@literal &}AI client for consuming data from A{@literal &}AI.
+     *
+     * @param aaiClientConfiguration - A{@literal &}AI client config
+     */
+    public AaiConsumerClient(AaiClientConfiguration aaiClientConfiguration) {
+        closeableHttpClient = new AaiClientImpl(aaiClientConfiguration).getAaiHttpClient();
+        aaiHost = aaiClientConfiguration.aaiHost();
+        aaiProtocol = aaiClientConfiguration.aaiProtocol();
+        aaiHostPortNumber = aaiClientConfiguration.aaiPort();
+        aaiPath = aaiClientConfiguration.aaiBasePath() + aaiClientConfiguration.aaiPnfPath();
+        aaiHeaders = aaiClientConfiguration.aaiHeaders();
+    }
+
+    /**
+     * Function which call http client for getting object from A{@literal &}AI.
+     *
+     * @param consumerDmaapModel - helper object for uri generation
+     * @return - status code of operation
+     * @throws IOException - Apache HTTP client exception
+     */
+    public Optional<String> getHttpResponse(ConsumerDmaapModel consumerDmaapModel) throws IOException {
+        Optional<HttpRequestBase> request = createRequest(consumerDmaapModel);
+        try {
+            return closeableHttpClient.execute(request.get(), aaiResponseHandler());
+        } catch (IOException e) {
+            logger.warn("Exception while executing http client: ", e);
+            throw new IOException();
+        }
+    }
+
+    private URI createAaiExtendedUri(String pnfName) {
+
+        URI extendedUri = null;
+
+        final URIBuilder uriBuilder = new URIBuilder()
+            .setScheme(aaiProtocol)
+            .setHost(aaiHost)
+            .setPort(aaiHostPortNumber)
+            .setPath(aaiPath + "/" + pnfName);
+
+        try {
+            extendedUri = uriBuilder.build();
+            logger.trace("Building extended URI: {}", extendedUri);
+        } catch (URISyntaxException e) {
+            logger.warn("Exception while building extended URI: {}", e);
+        }
+
+        return extendedUri;
+    }
+
+    private ResponseHandler<Optional<String>> aaiResponseHandler() {
+        return httpResponse -> {
+            final int responseCode = httpResponse.getStatusLine().getStatusCode();
+            logger.info("Status code of operation: {}", responseCode);
+            final HttpEntity responseEntity = httpResponse.getEntity();
+
+            if (HttpUtils.isSuccessfulResponseCode(responseCode)) {
+                logger.trace("HTTP response successful.");
+                final String aaiResponse = EntityUtils.toString(responseEntity);
+                return Optional.of(aaiResponse);
+            } else {
+                String aaiResponse = responseEntity != null ? EntityUtils.toString(responseEntity) : "";
+                logger.warn("HTTP response not successful : {}", aaiResponse);
+                return Optional.of(String.valueOf(responseCode));
+            }
+        };
+    }
+
+    private HttpRequestBase createHttpRequest(URI extendedUri) {
+        return isExtendedUriNotNull(extendedUri) ? new HttpGet(extendedUri) : null;
+    }
+
+    private Boolean isExtendedUriNotNull(URI extendedUri) {
+        return extendedUri != null;
+    }
+
+    private Optional<HttpRequestBase> createRequest(ConsumerDmaapModel consumerDmaapModel) {
+        final URI extendedUri = createAaiExtendedUri(consumerDmaapModel.getPnfName());
+        HttpRequestBase request = createHttpRequest(extendedUri);
+        aaiHeaders.forEach(Objects.requireNonNull(request)::addHeader);
+        Objects.requireNonNull(request).addHeader("Content-Type", "application/json");
+        return Optional.of(request);
+    }
+}
diff --git a/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiExtendedHttpClient.java b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiExtendedHttpClient.java
new file mode 100644
index 00000000..3c85b68e
--- /dev/null
+++ b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiExtendedHttpClient.java
@@ -0,0 +1,33 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Datafile Collector Service
+ * ================================================================================
+ * 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.collectors.datafile.service;
+
+import java.net.URISyntaxException;
+import java.util.Optional;
+
+import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
+
+
+@FunctionalInterface
+public interface AaiExtendedHttpClient {
+
+    Optional<Integer> getHttpResponse(ConsumerDmaapModel consumerDmaapModel) throws URISyntaxException;
+}
diff --git a/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiReactiveWebClient.java b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiReactiveWebClient.java
new file mode 100644
index 00000000..9a9ff27e
--- /dev/null
+++ b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/AaiReactiveWebClient.java
@@ -0,0 +1,85 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Datafile Collector Service
+ * ================================================================================
+ * 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.collectors.datafile.service;
+
+import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
+
+import java.util.Map;
+
+import org.onap.dcaegen2.collectors.datafile.config.AaiClientConfiguration;
+import org.onap.dcaegen2.collectors.datafile.service.AaiReactiveWebClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+
+public class AaiReactiveWebClient {
+
+    private Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    private String aaiUserName;
+    private String aaiUserPassword;
+    private Map<String, String> aaiHeaders;
+
+    /**
+     * Creating AaiReactiveWebClient.
+     * @param configuration - configuration object
+     * @return AaiReactiveWebClient
+     */
+    public AaiReactiveWebClient fromConfiguration(AaiClientConfiguration configuration) {
+        this.aaiUserName = configuration.aaiUserName();
+        this.aaiUserPassword = configuration.aaiUserPassword();
+        this.aaiHeaders = configuration.aaiHeaders();
+        return this;
+    }
+
+    /**
+     * Construct Reactive WebClient with appropriate settings.
+     *
+     * @return WebClient
+     */
+    public WebClient build() {
+        return WebClient.builder()
+                .defaultHeaders(httpHeaders -> httpHeaders.setAll(aaiHeaders))
+                .filter(basicAuthentication(aaiUserName, aaiUserPassword))
+                .filter(logRequest())
+                .filter(logResponse())
+                .build();
+    }
+
+    private 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);
+        });
+    }
+
+    private ExchangeFilterFunction logResponse() {
+        return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
+            logger.info("Response Status {}", clientResponse.statusCode());
+            return Mono.just(clientResponse);
+        });
+    }
+}
diff --git a/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/AaiProducerReactiveHttpClient.java b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/AaiProducerReactiveHttpClient.java
new file mode 100644
index 00000000..65ae40b8
--- /dev/null
+++ b/datafile-aai-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/AaiProducerReactiveHttpClient.java
@@ -0,0 +1,99 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Datafile Collector Service
+ * ================================================================================
+ * 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.collectors.datafile.service.producer;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+import org.apache.http.client.utils.URIBuilder;
+import org.onap.dcaegen2.collectors.datafile.config.AaiClientConfiguration;
+import org.onap.dcaegen2.collectors.datafile.exceptions.AaiRequestException;
+import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.service.producer.AaiProducerReactiveHttpClient;
+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;
+
+
+public class AaiProducerReactiveHttpClient {
+
+    private WebClient webClient;
+    private final String aaiHost;
+    private final String aaiProtocol;
+    private final Integer aaiHostPortNumber;
+    private final String aaiBasePath;
+
+
+    /**
+     * Constructor of AaiProducerReactiveHttpClient.
+     *
+     * @param configuration - AAI producer configuration object
+     */
+    public AaiProducerReactiveHttpClient(AaiClientConfiguration configuration) {
+        this.aaiHost = configuration.aaiHost();
+        this.aaiProtocol = configuration.aaiProtocol();
+        this.aaiHostPortNumber = configuration.aaiPort();
+        this.aaiBasePath = configuration.aaiBasePath();
+    }
+
+    /**
+     * Function for calling AAI Http producer - patch request to AAI database.
+     *
+     * @param consumerDmaapModelMono - object which will be sent to AAI database
+     * @return status code of operation
+     */
+    public Mono<Integer> getAaiProducerResponse(Mono<ConsumerDmaapModel> consumerDmaapModelMono) {
+        return consumerDmaapModelMono.flatMap(this::patchAaiRequest);
+    }
+
+    public AaiProducerReactiveHttpClient createAaiWebClient(WebClient webClient) {
+        this.webClient = webClient;
+        return this;
+    }
+
+    private Mono<Integer> patchAaiRequest(ConsumerDmaapModel dmaapModel) {
+        try {
+            return webClient.patch()
+                .uri(getUri(dmaapModel.getPnfName()))
+                .body(BodyInserters.fromObject(dmaapModel))
+                .retrieve()
+                .onStatus(
+                    HttpStatus::is4xxClientError,
+                    clientResponse -> Mono.error(new AaiRequestException("HTTP 400"))
+                )
+                .onStatus(HttpStatus::is5xxServerError,
+                    clientResponse -> Mono.error(new AaiRequestException("HTTP 500")))
+                .bodyToMono(Integer.class);
+        } catch (URISyntaxException e) {
+            return Mono.error(e);
+        }
+    }
+
+    URI getUri(String pnfName) throws URISyntaxException {
+        return new URIBuilder()
+            .setScheme(aaiProtocol)
+            .setHost(aaiHost)
+            .setPort(aaiHostPortNumber)
+            .setPath(aaiBasePath + "/" + pnfName)
+            .build();
+    }
+}
diff --git a/datafile-aai-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/AaiConsumerClientTest.java b/datafile-aai-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/AaiConsumerClientTest.java
new file mode 100644
index 00000000..719418fb
--- /dev/null
+++ b/datafile-aai-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/AaiConsumerClientTest.java
@@ -0,0 +1,95 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Datafile Collector Service
+ * ================================================================================
+ * 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.collectors.datafile.service;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import org.apache.http.client.ResponseHandler;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.collectors.datafile.config.AaiClientConfiguration;
+import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.service.AaiConsumerClient;
+
+class AaiConsumerClientTest {
+
+    private static AaiConsumerClient testedObject;
+    private static AaiClientConfiguration aaiHttpClientConfigurationMock = mock(AaiClientConfiguration.class);
+    private static CloseableHttpClient closeableHttpClientMock = mock(CloseableHttpClient.class);
+    private static final String JSON_MESSAGE = "{ \"pnf-id\": \"example-pnf-id-val-22343\", "
+        + "\"regional-resource-zone\":null, \"ipaddress-v4-oam\": \"11.22.33.44\" }";
+    private static ConsumerDmaapModel consumerDmaapModelMock = mock(ConsumerDmaapModel.class);
+    private static final String PNF_NAME = "nokia-pnf-nhfsadhff";
+
+    @BeforeAll
+    static void setup() throws NoSuchFieldException, IllegalAccessException {
+
+        Map<String, String> aaiHeaders = new HashMap<>();
+        aaiHeaders.put("X-FromAppId", "datafile");
+        aaiHeaders.put("X-TransactionId", "9999");
+        aaiHeaders.put("Accept", "application/json");
+        aaiHeaders.put("Authorization", "Basic QUFJOkFBSQ==");
+        aaiHeaders.put("Real-Time", "true");
+        aaiHeaders.put("Content-Type", "application/json");
+
+        when(aaiHttpClientConfigurationMock.aaiHost()).thenReturn("54.45.33.2");
+        when(aaiHttpClientConfigurationMock.aaiProtocol()).thenReturn("https");
+        when(aaiHttpClientConfigurationMock.aaiPort()).thenReturn(1234);
+        when(aaiHttpClientConfigurationMock.aaiUserName()).thenReturn("Datafile");
+        when(aaiHttpClientConfigurationMock.aaiUserPassword()).thenReturn("Datafile");
+        when(aaiHttpClientConfigurationMock.aaiBasePath()).thenReturn("/aai/v11");
+        when(aaiHttpClientConfigurationMock.aaiPnfPath()).thenReturn("/network/pnfs/pnf");
+        when(aaiHttpClientConfigurationMock.aaiHeaders()).thenReturn(aaiHeaders);
+
+        when(consumerDmaapModelMock.getPnfName()).thenReturn(PNF_NAME);
+
+        testedObject = new AaiConsumerClient(aaiHttpClientConfigurationMock);
+        setField();
+    }
+
+
+    @Test
+    void getExtendedDetails_returnsSuccess() throws IOException {
+
+        when(closeableHttpClientMock.execute(any(HttpGet.class), any(ResponseHandler.class)))
+            .thenReturn(Optional.of(JSON_MESSAGE));
+        Optional<String> actualResult = testedObject.getHttpResponse(consumerDmaapModelMock);
+        Assertions.assertEquals(Optional.of(JSON_MESSAGE), actualResult);
+    }
+
+
+    private static void setField() throws NoSuchFieldException, IllegalAccessException {
+        Field field = testedObject.getClass().getDeclaredField("closeableHttpClient");
+        field.setAccessible(true);
+        field.set(testedObject, closeableHttpClientMock);
+    }
+}
diff --git a/datafile-aai-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/AaiHttpClientImplTest.java b/datafile-aai-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/AaiHttpClientImplTest.java
new file mode 100644
index 00000000..dcf29549
--- /dev/null
+++ b/datafile-aai-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/AaiHttpClientImplTest.java
@@ -0,0 +1,55 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Datafile Collector Service
+ * ================================================================================
+ * 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.collectors.datafile.service;
+
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.collectors.datafile.config.AaiClientConfiguration;
+import org.onap.dcaegen2.collectors.datafile.service.AaiClientImpl;
+
+class AaiHttpClientImplTest {
+
+    private static AaiClientImpl testedObject;
+
+
+    @BeforeAll
+    public static void setup() {
+        AaiClientConfiguration aaiHttpClientConfigurationMock = mock(AaiClientConfiguration.class);
+        when(aaiHttpClientConfigurationMock.aaiHost()).thenReturn("54.45.33.2");
+        when(aaiHttpClientConfigurationMock.aaiProtocol()).thenReturn("https");
+        when(aaiHttpClientConfigurationMock.aaiPort()).thenReturn(1234);
+        when(aaiHttpClientConfigurationMock.aaiUserName()).thenReturn("PNF");
+        when(aaiHttpClientConfigurationMock.aaiUserPassword()).thenReturn("PNF");
+        when(aaiHttpClientConfigurationMock.aaiIgnoreSslCertificateErrors()).thenReturn(true);
+
+        testedObject  = new AaiClientImpl(aaiHttpClientConfigurationMock);
+    }
+
+    @Test
+    public void getAaiHttpClientObject_shouldNotBeNull() {
+        assertNotNull(testedObject.getAaiHttpClient());
+    }
+}
+
diff --git a/datafile-aai-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/AaiProducerReactiveHttpClientTest.java b/datafile-aai-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/AaiProducerReactiveHttpClientTest.java
new file mode 100644
index 00000000..65f13553
--- /dev/null
+++ b/datafile-aai-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/AaiProducerReactiveHttpClientTest.java
@@ -0,0 +1,142 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Datafile Collector Service
+ * ================================================================================
+ * 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.collectors.datafile.service.producer;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.collectors.datafile.config.AaiClientConfiguration;
+import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModelForUnitTest;
+import org.onap.dcaegen2.collectors.datafile.service.producer.AaiProducerReactiveHttpClient;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClient.ResponseSpec;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+
+class AaiProducerReactiveHttpClientTest {
+
+    private static AaiProducerReactiveHttpClient aaiProducerReactiveHttpClient;
+
+    private static final Integer SUCCESS_RESPONSE = 200;
+
+    private static AaiClientConfiguration aaiConfigurationMock = mock(AaiClientConfiguration.class);
+    private static WebClient webClient = mock(WebClient.class);
+
+    private static ConsumerDmaapModel dmaapModel = new ConsumerDmaapModelForUnitTest();
+    private static WebClient.RequestBodyUriSpec requestBodyUriSpec;
+    private static ResponseSpec responseSpec;
+
+    private static Map<String, String> aaiHeaders;
+
+    @BeforeAll
+    static void setUp() {
+        setupHeaders();
+
+        when(aaiConfigurationMock.aaiHost()).thenReturn("54.45.33.2");
+        when(aaiConfigurationMock.aaiProtocol()).thenReturn("https");
+        when(aaiConfigurationMock.aaiPort()).thenReturn(1234);
+        when(aaiConfigurationMock.aaiUserName()).thenReturn("Datafile");
+        when(aaiConfigurationMock.aaiUserPassword()).thenReturn("Datafile");
+        when(aaiConfigurationMock.aaiBasePath()).thenReturn("/aai/v11");
+        when(aaiConfigurationMock.aaiPnfPath()).thenReturn("/network/pnfs/pnf");
+        when(aaiConfigurationMock.aaiHeaders()).thenReturn(aaiHeaders);
+
+        aaiProducerReactiveHttpClient = new AaiProducerReactiveHttpClient(aaiConfigurationMock);
+
+        webClient = spy(WebClient.builder()
+            .defaultHeaders(httpHeaders -> httpHeaders.setAll(aaiHeaders))
+            .filter(basicAuthentication(aaiConfigurationMock.aaiUserName(), aaiConfigurationMock.aaiUserPassword()))
+            .build());
+
+        requestBodyUriSpec = mock(WebClient.RequestBodyUriSpec.class);
+        responseSpec = mock(ResponseSpec.class);
+    }
+
+
+    @Test
+    void getAaiProducerResponse_shouldReturn200() {
+        //given
+        Mono<Integer> expectedResult = Mono.just(SUCCESS_RESPONSE);
+
+        //when
+        mockWebClientDependantObject();
+        doReturn(expectedResult).when(responseSpec).bodyToMono(Integer.class);
+        aaiProducerReactiveHttpClient.createAaiWebClient(webClient);
+        Mono<Integer> response = aaiProducerReactiveHttpClient.getAaiProducerResponse(Mono.just(dmaapModel));
+
+        //then
+        StepVerifier.create(response).expectSubscription()
+            .expectNextMatches(results -> {
+                Assertions.assertEquals(results, expectedResult.block());
+                return true;
+            }).verifyComplete();
+    }
+
+    @Test
+    void getHttpResponse_whenUriSyntaxExceptionHasBeenThrown() throws URISyntaxException {
+        ///given
+        aaiProducerReactiveHttpClient = spy(aaiProducerReactiveHttpClient);
+        //when
+        when(webClient.patch()).thenReturn(requestBodyUriSpec);
+        aaiProducerReactiveHttpClient.createAaiWebClient(webClient);
+        when(aaiProducerReactiveHttpClient.getUri("pnfName")).thenThrow(URISyntaxException.class);
+
+        //then
+        StepVerifier.create(
+            aaiProducerReactiveHttpClient.getAaiProducerResponse(
+                Mono.just(dmaapModel)
+            )).expectSubscription().expectError(Exception.class).verify();
+    }
+
+
+    private void mockWebClientDependantObject() {
+        WebClient.RequestHeadersSpec requestHeadersSpec = mock(WebClient.RequestHeadersSpec.class);
+        when(webClient.patch()).thenReturn(requestBodyUriSpec);
+        when(requestBodyUriSpec.uri((URI) any())).thenReturn(requestBodyUriSpec);
+        when(requestBodyUriSpec.body(any())).thenReturn(requestHeadersSpec);
+        doReturn(responseSpec).when(requestHeadersSpec).retrieve();
+        doReturn(responseSpec).when(responseSpec).onStatus(any(), any());
+    }
+
+    private static void setupHeaders() {
+        aaiHeaders = new HashMap<>();
+        aaiHeaders.put("X-FromAppId", "Datafile");
+        aaiHeaders.put("X-TransactionId", "vv-temp");
+        aaiHeaders.put("Accept", "application/json");
+        aaiHeaders.put("Real-Time", "true");
+        aaiHeaders.put("Content-Type", "application/merge-patch+json");
+    }
+
+}
+