summaryrefslogtreecommitdiffstats
path: root/prh-app-server/src/main/java
diff options
context:
space:
mode:
authorwasala <przemyslaw.wasala@nokia.com>2018-05-08 13:48:44 +0200
committerwasala <przemyslaw.wasala@nokia.com>2018-05-09 14:20:35 +0200
commit44ce56cf2e6bfae17c443901ac6ded8ba8cdd414 (patch)
treed2700e58cff23035f8f489ef7579ecee04ef3cc6 /prh-app-server/src/main/java
parent7e7aa0c165665f74b628ba5c95fb3cdae72f4449 (diff)
Added dmaap consumer task execution to workflow
Change-Id: I67904a8ffb7d401fa2188fb046e28474eded4377 Issue-ID: DCAEGEN2-474 Signed-off-by: wasala <przemyslaw.wasala@nokia.com>
Diffstat (limited to 'prh-app-server/src/main/java')
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AppConfig.java94
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/PrhAppConfig.java33
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/HeartbeatController.java15
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleController.java25
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/exceptions/PrhTaskException.java2
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/model/ConsumerDmaapModel.java43
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java92
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIConsumerTaskImpl.java28
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTask.java4
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTaskImpl.java39
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java4
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java34
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java5
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java31
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java13
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/Task.java6
16 files changed, 269 insertions, 199 deletions
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AppConfig.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AppConfig.java
index 15db50e1..fba8985f 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AppConfig.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AppConfig.java
@@ -109,63 +109,63 @@ public class AppConfig extends PrhAppConfig {
@Override
public DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
return new ImmutableDmaapConsumerConfiguration.Builder()
- .dmaapUserPassword(
- Optional.ofNullable(consumerDmaapUserPassword).orElse(dmaapConsumerConfiguration.dmaapUserPassword()))
- .dmaapUserName(
- Optional.ofNullable(consumerDmaapUserName).orElse(dmaapConsumerConfiguration.dmaapUserName()))
- .dmaapHostName(
- Optional.ofNullable(consumerDmaapHostName).orElse(dmaapConsumerConfiguration.dmaapHostName()))
- .dmaapPortNumber(
- Optional.ofNullable(consumerDmaapPortNumber).orElse(dmaapConsumerConfiguration.dmaapPortNumber()))
- .dmaapProtocol(
- Optional.ofNullable(consumerDmaapProtocol).orElse(dmaapConsumerConfiguration.dmaapProtocol()))
- .dmaapContentType(
- Optional.ofNullable(consumerDmaapContentType).orElse(dmaapConsumerConfiguration.dmaapContentType()))
- .dmaapTopicName(
- Optional.ofNullable(consumerDmaapTopicName).orElse(dmaapConsumerConfiguration.dmaapTopicName()))
- .messageLimit(
- Optional.ofNullable(consumerMessageLimit).orElse(dmaapConsumerConfiguration.messageLimit()))
- .timeoutMS(Optional.ofNullable(consumerTimeoutMS).orElse(dmaapConsumerConfiguration.timeoutMS()))
- .consumerGroup(Optional.ofNullable(consumerGroup).orElse(dmaapConsumerConfiguration.consumerGroup()))
- .consumerId(Optional.ofNullable(consumerId).orElse(dmaapConsumerConfiguration.consumerId()))
- .build();
+ .dmaapUserPassword(
+ Optional.ofNullable(consumerDmaapUserPassword).orElse(dmaapConsumerConfiguration.dmaapUserPassword()))
+ .dmaapUserName(
+ Optional.ofNullable(consumerDmaapUserName).orElse(dmaapConsumerConfiguration.dmaapUserName()))
+ .dmaapHostName(
+ Optional.ofNullable(consumerDmaapHostName).orElse(dmaapConsumerConfiguration.dmaapHostName()))
+ .dmaapPortNumber(
+ Optional.ofNullable(consumerDmaapPortNumber).orElse(dmaapConsumerConfiguration.dmaapPortNumber()))
+ .dmaapProtocol(
+ Optional.ofNullable(consumerDmaapProtocol).orElse(dmaapConsumerConfiguration.dmaapProtocol()))
+ .dmaapContentType(
+ Optional.ofNullable(consumerDmaapContentType).orElse(dmaapConsumerConfiguration.dmaapContentType()))
+ .dmaapTopicName(
+ Optional.ofNullable(consumerDmaapTopicName).orElse(dmaapConsumerConfiguration.dmaapTopicName()))
+ .messageLimit(
+ Optional.ofNullable(consumerMessageLimit).orElse(dmaapConsumerConfiguration.messageLimit()))
+ .timeoutMS(Optional.ofNullable(consumerTimeoutMS).orElse(dmaapConsumerConfiguration.timeoutMS()))
+ .consumerGroup(Optional.ofNullable(consumerGroup).orElse(dmaapConsumerConfiguration.consumerGroup()))
+ .consumerId(Optional.ofNullable(consumerId).orElse(dmaapConsumerConfiguration.consumerId()))
+ .build();
}
@Override
public AAIClientConfiguration getAAIClientConfiguration() {
return new ImmutableAAIClientConfiguration.Builder()
- .aaiHost(Optional.ofNullable(aaiHost).orElse(aaiClientConfiguration.aaiHost()))
- .aaiHostPortNumber(
- Optional.ofNullable(aaiHostPortNumber).orElse(aaiClientConfiguration.aaiHostPortNumber()))
- .aaiIgnoreSSLCertificateErrors(
- Optional.ofNullable(aaiIgnoreSSLCertificateErrors)
- .orElse(aaiClientConfiguration.aaiIgnoreSSLCertificateErrors()))
- .aaiProtocol(Optional.ofNullable(aaiProtocol).orElse(aaiClientConfiguration.aaiProtocol()))
- .aaiUserName(
- Optional.ofNullable(aaiUserName).orElse(aaiClientConfiguration.aaiUserName()))
- .aaiUserPassword(Optional.ofNullable(
- aaiUserPassword).orElse(aaiClientConfiguration.aaiUserPassword()))
- .build();
+ .aaiHost(Optional.ofNullable(aaiHost).orElse(aaiClientConfiguration.aaiHost()))
+ .aaiHostPortNumber(
+ Optional.ofNullable(aaiHostPortNumber).orElse(aaiClientConfiguration.aaiHostPortNumber()))
+ .aaiIgnoreSSLCertificateErrors(
+ Optional.ofNullable(aaiIgnoreSSLCertificateErrors)
+ .orElse(aaiClientConfiguration.aaiIgnoreSSLCertificateErrors()))
+ .aaiProtocol(Optional.ofNullable(aaiProtocol).orElse(aaiClientConfiguration.aaiProtocol()))
+ .aaiUserName(
+ Optional.ofNullable(aaiUserName).orElse(aaiClientConfiguration.aaiUserName()))
+ .aaiUserPassword(Optional.ofNullable(
+ aaiUserPassword).orElse(aaiClientConfiguration.aaiUserPassword()))
+ .build();
}
@Override
public DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
return new ImmutableDmaapPublisherConfiguration.Builder()
- .dmaapContentType(
- Optional.ofNullable(producerDmaapContentType).orElse(dmaapPublisherConfiguration.dmaapContentType()))
- .dmaapHostName(
- Optional.ofNullable(producerDmaapHostName).orElse(dmaapPublisherConfiguration.dmaapHostName()))
- .dmaapPortNumber(
- Optional.ofNullable(producerDmaapPortNumber).orElse(dmaapPublisherConfiguration.dmaapPortNumber()))
- .dmaapProtocol(
- Optional.ofNullable(producerDmaapProtocol).orElse(dmaapPublisherConfiguration.dmaapProtocol()))
- .dmaapTopicName(
- Optional.ofNullable(producerDmaapTopicName).orElse(dmaapPublisherConfiguration.dmaapTopicName()))
- .dmaapUserName(
- Optional.ofNullable(producerDmaapUserName).orElse(dmaapPublisherConfiguration.dmaapUserName()))
- .dmaapUserPassword(
- Optional.ofNullable(producerDmaapUserPassword).orElse(dmaapPublisherConfiguration.dmaapUserPassword()))
- .build();
+ .dmaapContentType(
+ Optional.ofNullable(producerDmaapContentType).orElse(dmaapPublisherConfiguration.dmaapContentType()))
+ .dmaapHostName(
+ Optional.ofNullable(producerDmaapHostName).orElse(dmaapPublisherConfiguration.dmaapHostName()))
+ .dmaapPortNumber(
+ Optional.ofNullable(producerDmaapPortNumber).orElse(dmaapPublisherConfiguration.dmaapPortNumber()))
+ .dmaapProtocol(
+ Optional.ofNullable(producerDmaapProtocol).orElse(dmaapPublisherConfiguration.dmaapProtocol()))
+ .dmaapTopicName(
+ Optional.ofNullable(producerDmaapTopicName).orElse(dmaapPublisherConfiguration.dmaapTopicName()))
+ .dmaapUserName(
+ Optional.ofNullable(producerDmaapUserName).orElse(dmaapPublisherConfiguration.dmaapUserName()))
+ .dmaapUserPassword(
+ Optional.ofNullable(producerDmaapUserPassword).orElse(dmaapPublisherConfiguration.dmaapUserPassword()))
+ .build();
}
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/PrhAppConfig.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/PrhAppConfig.java
index 0473c1aa..3df25d00 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/PrhAppConfig.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/PrhAppConfig.java
@@ -52,7 +52,6 @@ public abstract class PrhAppConfig implements Config {
private static final String DMAAP_CONSUMER = "dmaapConsumerConfiguration";
private static final Logger logger = LoggerFactory.getLogger(PrhAppConfig.class);
- private static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
AAIClientConfiguration aaiClientConfiguration;
@@ -91,40 +90,28 @@ public abstract class PrhAppConfig implements Config {
if (rootElement.isJsonObject()) {
jsonObject = rootElement.getAsJsonObject();
aaiClientConfiguration = deserializeType(gsonBuilder,
- jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(AAI).getAsJsonObject(AAI_CONFIG),
- AAIClientConfiguration.class);
+ jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(AAI).getAsJsonObject(AAI_CONFIG),
+ AAIClientConfiguration.class);
dmaapConsumerConfiguration = deserializeType(gsonBuilder,
- jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP).getAsJsonObject(DMAAP_CONSUMER),
- DmaapConsumerConfiguration.class);
+ jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP).getAsJsonObject(DMAAP_CONSUMER),
+ DmaapConsumerConfiguration.class);
dmaapPublisherConfiguration = deserializeType(gsonBuilder,
- jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP).getAsJsonObject(DMAAP_PRODUCER),
- DmaapPublisherConfiguration.class);
+ jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP).getAsJsonObject(DMAAP_PRODUCER),
+ DmaapPublisherConfiguration.class);
}
} catch (FileNotFoundException e) {
- logger
- .error(
- "Configuration PrhAppConfig initFileStreamReader()::FileNotFoundException :: Execution Time - {}:{}",
- dateTimeFormatter.format(
- LocalDateTime.now()), e);
+ logger.warn("File doesn't exist in filepath: {}", filepath, e);
} catch (IOException e) {
- logger
- .error(
- "Configuration PrhAppConfig initFileStreamReader()::IOException :: Execution Time - {}:{}",
- dateTimeFormatter.format(
- LocalDateTime.now()), e);
+ logger.warn("Problem with file loading, file: {}", filepath, e);
} catch (JsonSyntaxException e) {
- logger
- .error(
- "Configuration PrhAppConfig initFileStreamReader()::JsonSyntaxException :: Execution Time - {}:{}",
- dateTimeFormatter.format(
- LocalDateTime.now()), e);
+ logger.warn("Problem with Json deserialization", e);
}
}
private <T> T deserializeType(@NotNull GsonBuilder gsonBuilder, @NotNull JsonObject jsonObject,
- @NotNull Class<T> type) {
+ @NotNull Class<T> type) {
return gsonBuilder.create().fromJson(jsonObject, type);
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/HeartbeatController.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/HeartbeatController.java
index 05b91840..275685fe 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/HeartbeatController.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/HeartbeatController.java
@@ -44,7 +44,6 @@ import reactor.core.publisher.Mono;
public class HeartbeatController {
private static final Logger logger = LoggerFactory.getLogger(PrhAppConfig.class);
- private static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
@RequestMapping(value = "heartbeat", method = RequestMethod.GET)
@ApiOperation(value = "Returns liveness of PRH service")
@@ -56,15 +55,9 @@ public class HeartbeatController {
}
)
public Mono<ResponseEntity<String>> heartbeat() {
- logger.debug("Receiving request on on thread={} , time={} ", Thread.currentThread().getName(),
- dateTimeFormatter.format(
- LocalDateTime.now()));
-
- return Mono.defer(() -> {
- logger.debug("Sending response on thread={} , time={} ", Thread.currentThread().getName(),
- dateTimeFormatter.format(
- LocalDateTime.now()));
- return Mono.just(new ResponseEntity<>("I'm living", HttpStatus.OK));
- });
+ logger.trace("Receiving heartbeat request");
+ return Mono.defer(() ->
+ Mono.just(new ResponseEntity<>("I'm living", HttpStatus.OK))
+ );
}
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleController.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleController.java
index 60572cb3..4cde51bf 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleController.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleController.java
@@ -50,7 +50,6 @@ import reactor.core.publisher.Mono;
public class ScheduleController {
private static final Logger logger = LoggerFactory.getLogger(PrhAppConfig.class);
- private static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
private static final int SCHEDULING_DELAY = 20000;
private static volatile List<ScheduledFuture> scheduledFutureList = new ArrayList<>();
@@ -67,14 +66,14 @@ public class ScheduleController {
@RequestMapping(value = "start", method = RequestMethod.GET)
@ApiOperation(value = "Start scheduling worker request")
public Mono<ResponseEntity<String>> startTasks() {
- logDebug("Starting scheduling worker request on on thread={} , time={} ");
+ logger.trace("Receiving start scheduling worker request");
return Mono.fromSupplier(this::tryToStartTask).map(this::createStartTaskResponse);
}
@RequestMapping(value = "stopPrh", method = RequestMethod.GET)
- @ApiOperation(value = "Stop scheduling worker request")
+ @ApiOperation(value = "Receiving stop scheduling worker request")
public Mono<ResponseEntity<String>> stopTask() {
- logDebug("Stopping scheduling worker request on on thread={} , time={} ");
+ logger.trace("Receiving stop scheduling worker request");
return getResponseFromCancellationOfTasks();
}
@@ -82,10 +81,9 @@ public class ScheduleController {
private synchronized Mono<ResponseEntity<String>> getResponseFromCancellationOfTasks() {
scheduledFutureList.forEach(x -> x.cancel(false));
scheduledFutureList.clear();
- return Mono.defer(() -> {
- logDebug("Sending success response on stopping task execution thread={} , time={} ");
- return Mono.just(new ResponseEntity<>("PRH Service has already been stopped!", HttpStatus.CREATED));
- });
+ return Mono.defer(() ->
+ Mono.just(new ResponseEntity<>("PRH Service has already been stopped!", HttpStatus.CREATED))
+ );
}
@ApiOperation(value = "Start task if possible")
@@ -103,20 +101,9 @@ public class ScheduleController {
@ApiOperation(value = "Sends success or error response on starting task execution")
private ResponseEntity<String> createStartTaskResponse(boolean wasScheduled) {
if (wasScheduled) {
- logDebug("Sending success response on starting task execution thread={} , time={} ");
return new ResponseEntity<>("PRH Service has been started!", HttpStatus.CREATED);
} else {
- logDebug("Sending error response on starting task execution thread={} , time={} ");
return new ResponseEntity<>("PRH Service is still running!", HttpStatus.NOT_ACCEPTABLE);
}
}
-
- private static void logDebug(String message) {
- if (logger.isDebugEnabled()) {
- logger.debug(message,
- Thread.currentThread().getName(),
- dateTimeFormatter.format(
- LocalDateTime.now()));
- }
- }
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/exceptions/PrhTaskException.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/exceptions/PrhTaskException.java
index 33193609..646ca7e7 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/exceptions/PrhTaskException.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/exceptions/PrhTaskException.java
@@ -24,7 +24,7 @@ package org.onap.dcaegen2.services.prh.exceptions;
*/
public class PrhTaskException extends Exception {
- PrhTaskException(String message) {
+ public PrhTaskException(String message) {
super(message);
}
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/model/ConsumerDmaapModel.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/model/ConsumerDmaapModel.java
new file mode 100644
index 00000000..a3c7d1d3
--- /dev/null
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/model/ConsumerDmaapModel.java
@@ -0,0 +1,43 @@
+/*
+ * ============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.model;
+
+import com.google.gson.annotations.SerializedName;
+import org.immutables.gson.Gson;
+import org.immutables.value.Value;
+
+/**
+ * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/8/18
+ */
+
+@Value.Immutable(prehash = true)
+@Gson.TypeAdapters
+public interface ConsumerDmaapModel {
+
+ @SerializedName("pnf-name")
+ String getPnfName();
+
+ @SerializedName("ipaddress-v4-oam")
+ String getIpv4();
+
+ @SerializedName("ipaddress-v6-oam")
+ String getIpv6();
+
+}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java
new file mode 100644
index 00000000..e2d9816e
--- /dev/null
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java
@@ -0,0 +1,92 @@
+/*
+ * ============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 com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import java.util.Optional;
+import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
+import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
+
+/**
+ * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/8/18
+ */
+public class DmaapConsumerJsonParser {
+
+ private static final String EVENT = "event";
+ private static final String OTHER_FIELDS = "otherFields";
+ private static final String PNF_OAM_IPV_4_ADDRESS = "pnfOamIpv4Address";
+ private static final String PNF_OAM_IPV_6_ADDRESS = "pnfOamIpv6Address";
+ private static final String PNF_VENDOR_NAME = "pnfVendorName";
+ private static final String PNF_SERIAL_NUMBER = "pnfSerialNumber";
+
+
+ private DmaapConsumerJsonParser() {
+ }
+
+ public static ConsumerDmaapModel getJsonObject(String message) throws DmaapNotFoundException {
+ JsonElement jsonElement = new JsonParser().parse(message);
+ JsonObject jsonObject = jsonElement.getAsJsonObject();
+ return create(jsonObject);
+ }
+
+ private static ConsumerDmaapModel create(JsonObject jsonObject) throws DmaapNotFoundException {
+ if (containsHeader(jsonObject)) {
+ jsonObject = jsonObject.getAsJsonObject(EVENT).getAsJsonObject(OTHER_FIELDS);
+ String pnfVendorName = getValueFromJson(jsonObject, PNF_VENDOR_NAME);
+ String pnfSerialNumber = getValueFromJson(jsonObject, PNF_SERIAL_NUMBER);
+ String pnfOamIpv4Address = getValueFromJson(jsonObject, PNF_OAM_IPV_4_ADDRESS);
+ String pnfOamIpv6Address = getValueFromJson(jsonObject, PNF_OAM_IPV_6_ADDRESS);
+ if (isVendorAndSerialNotEmpty(pnfSerialNumber, pnfVendorName) && isIpPropertiesNotEmpty(pnfOamIpv4Address,
+ pnfOamIpv6Address)) {
+ String correlationID = pnfVendorName.substring(0, Math.min(pnfVendorName.length(), 3)).toUpperCase()
+ .concat(pnfSerialNumber);
+ return ImmutableConsumerDmaapModel.builder().pnfName(correlationID).ipv4(pnfOamIpv4Address)
+ .ipv6(pnfOamIpv6Address).build();
+ }
+ throw new DmaapNotFoundException("IPV4 and IPV6 are empty");
+ }
+ throw new DmaapNotFoundException("Incorrect JsonObject");
+ }
+
+ private static String getValueFromJson(JsonObject jsonObject, String jsonKey) {
+ if (jsonObject.has(jsonKey)) {
+ return jsonObject.get(jsonKey).getAsString();
+ }
+ return "";
+ }
+
+ private static boolean isVendorAndSerialNotEmpty(String pnfSerialNumber, String pnfVendorName) {
+ return ((pnfSerialNumber != null && !pnfSerialNumber.isEmpty()) && (pnfVendorName != null && !pnfVendorName
+ .isEmpty()));
+ }
+
+ private static boolean isIpPropertiesNotEmpty(String ipv4, String ipv6) {
+ return (ipv4 != null && !ipv4.isEmpty()) || (ipv6 != null
+ && !ipv6.isEmpty());
+ }
+
+ private static boolean containsHeader(JsonObject jsonObject) {
+ return jsonObject.has(EVENT) && jsonObject.getAsJsonObject(EVENT).has(OTHER_FIELDS);
+ }
+
+}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIConsumerTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIConsumerTaskImpl.java
index 3504bf98..1c80d3f9 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIConsumerTaskImpl.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIConsumerTaskImpl.java
@@ -40,46 +40,32 @@ import java.util.Optional;
public class AAIConsumerTaskImpl extends AAIConsumerTask<AAIClientConfiguration> {
private static final Logger logger = LoggerFactory.getLogger(ScheduledTasks.class);
- private static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
private final Config prhAppConfig;
- private final HttpRequestDetails requestDetails;
- private AAIProducerClient producerClient;
- Optional<String> response;
+ private Optional<String> response;
@Autowired
- public AAIConsumerTaskImpl(AppConfig prhAppConfig, HttpRequestDetails requestDetails) {
+ public AAIConsumerTaskImpl(AppConfig prhAppConfig) {
this.prhAppConfig = prhAppConfig;
- this.requestDetails = requestDetails;
}
@Override
protected void consume() throws AAINotFoundException {
- logger.debug("Start task AAIConsumerTask::publish() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
-
-
- producerClient = new AAIProducerClient(prhAppConfig.getAAIClientConfiguration());
-
- response = producerClient.getHttpResponse(requestDetails);
-
- logger.debug("End task AAIConsumerTask::publish() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
+ //TODO: @Piotr Wielebski
+ AAIProducerClient producerClient = new AAIProducerClient(prhAppConfig.getAAIClientConfiguration());
}
@Override
- public ResponseEntity execute(Object object) throws AAINotFoundException {
- logger.debug("Start task AAIProducerTaskImpl::execute() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
+ public Object execute(Object object) throws AAINotFoundException {
consume();
- logger.debug("End task AAIPublisherTaskImpl::execute() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
+ //TODO: @Piotr Wielebski
return null;
}
@Override
void initConfigs() {
+ logger.trace("initConfigs for AAIConsumerTaskImpl not needed/supported");
}
@Override
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTask.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTask.java
index 787e614d..8545949f 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTask.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTask.java
@@ -24,9 +24,9 @@ import org.onap.dcaegen2.services.prh.exceptions.AAINotFoundException;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
*/
-public abstract class AAIProducerTask<T> extends Task {
+public abstract class AAIProducerTask<T, U, V> extends Task {
- protected abstract void publish() throws AAINotFoundException;
+ protected abstract V publish(U message) throws AAINotFoundException;
protected abstract T resolveConfiguration();
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTaskImpl.java
index 5e8dcea1..a254890f 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTaskImpl.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTaskImpl.java
@@ -19,10 +19,13 @@
*/
package org.onap.dcaegen2.services.prh.tasks;
+import com.google.gson.Gson;
import org.onap.dcaegen2.services.config.AAIClientConfiguration;
import org.onap.dcaegen2.services.prh.configuration.AppConfig;
import org.onap.dcaegen2.services.prh.configuration.Config;
import org.onap.dcaegen2.services.prh.exceptions.AAINotFoundException;
+import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
import org.onap.dcaegen2.services.service.AAIProducerClient;
import org.onap.dcaegen2.services.utils.HttpRequestDetails;
import org.slf4j.Logger;
@@ -39,48 +42,38 @@ import java.util.Optional;
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
*/
@Component
-public class AAIProducerTaskImpl extends AAIProducerTask<AAIClientConfiguration> {
+public class AAIProducerTaskImpl extends AAIProducerTask<AAIClientConfiguration, ConsumerDmaapModel, Object> {
private static final Logger logger = LoggerFactory.getLogger(ScheduledTasks.class);
- private static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
private final Config prhAppConfig;
private AAIProducerClient producerClient;
- private HttpRequestDetails requestDetails;
- Optional<String> response;
+ private Optional<String> response;
@Autowired
- public AAIProducerTaskImpl(AppConfig prhAppConfig, HttpRequestDetails requestDetails) {
+ public AAIProducerTaskImpl(AppConfig prhAppConfig) {
this.prhAppConfig = prhAppConfig;
- this.requestDetails = requestDetails;
}
@Override
- protected void publish() throws AAINotFoundException {
- logger.debug("Start task AAIConsumerTask::publish() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
-
+ protected Object publish(ConsumerDmaapModel consumerDmaapModel) throws AAINotFoundException {
+ logger.trace("Method %M called with arg {}", consumerDmaapModel);
producerClient = new AAIProducerClient(prhAppConfig.getAAIClientConfiguration());
-
- response = producerClient.getHttpResponse(requestDetails);
-
- logger.debug("End task AAIConsumerTask::publish() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
-
+ //TODO: @Piotr Wielebski
+ response = producerClient.getHttpResponse(null);
+ return response.get();
}
@Override
- public ResponseEntity execute(Object object) throws AAINotFoundException {
- logger.debug("Start task AAIProducerTaskImpl::execute() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
- publish();
- logger.debug("End task AAIPublisherTaskImpl::execute() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
- return null;
+ public Object execute(Object object) throws AAINotFoundException {
+ logger.trace("Method %M called with arg {}", object);
+ //TODO: @Piotr Wielebski
+ return publish((ConsumerDmaapModel) object);
}
@Override
void initConfigs() {
+ logger.trace("initConfigs for AAIProducerTaskImpl not needed/supported");
}
@Override
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java
index 68a79e01..3e827075 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java
@@ -24,9 +24,9 @@ import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
*/
-public abstract class DmaapConsumerTask<T> extends Task {
+public abstract class DmaapConsumerTask<T, U, V> extends Task {
- protected abstract void consume() throws DmaapNotFoundException;
+ protected abstract V consume(U message) throws DmaapNotFoundException;
protected abstract T resolveConfiguration();
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java
index eb66361d..a2129fcd 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java
@@ -1,4 +1,4 @@
-/*-
+/*
* ============LICENSE_START=======================================================
* PNF-REGISTRATION-HANDLER
* ================================================================================
@@ -22,10 +22,14 @@ package org.onap.dcaegen2.services.prh.tasks;
import org.onap.dcaegen2.services.config.DmaapConsumerConfiguration;
import org.onap.dcaegen2.services.prh.configuration.AppConfig;
import org.onap.dcaegen2.services.prh.configuration.Config;
+import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
+import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
+import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser;
+import org.onap.dcaegen2.services.service.consumer.ExtendedDmaapConsumerHttpClientImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
@@ -35,11 +39,10 @@ import java.time.format.DateTimeFormatter;
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
*/
@Component
-public class DmaapConsumerTaskImpl extends DmaapConsumerTask<DmaapConsumerConfiguration> {
+public class DmaapConsumerTaskImpl extends DmaapConsumerTask<DmaapConsumerConfiguration, String, ConsumerDmaapModel> {
private static final Logger logger = LoggerFactory.getLogger(DmaapConsumerTaskImpl.class);
- private static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
private final Config prhAppConfig;
@@ -49,23 +52,18 @@ public class DmaapConsumerTaskImpl extends DmaapConsumerTask<DmaapConsumerConfig
}
@Override
- protected void consume() {
- logger.debug("Start task DmaapConsumerTask::consume() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
-
- logger.debug("End task DmaapConsumerTask::consume() :: Execution Time - {}",
- dateTimeFormatter.format(LocalDateTime.now()));
-
+ protected ConsumerDmaapModel consume(String message) throws DmaapNotFoundException {
+ logger.trace("Method %M called with arg {}", message);
+ return DmaapConsumerJsonParser.getJsonObject(message);
}
@Override
- public ResponseEntity execute(Object object) {
- logger.debug("Start task DmaapConsumerTask::execute() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
- consume();
- logger.debug("End task DmaapConsumerTask::execute() :: Execution Time - {}",
- dateTimeFormatter.format(LocalDateTime.now()));
- return null;
+ public Object execute(Object object) throws PrhTaskException {
+ logger.trace("Method %M called with arg {}", object);
+ ExtendedDmaapConsumerHttpClientImpl dmaapConsumerHttpClient = new ExtendedDmaapConsumerHttpClientImpl(
+ resolveConfiguration());
+ return consume((dmaapConsumerHttpClient.getHttpConsumerResponse().orElseThrow(() ->
+ new PrhTaskException("DmaapConsumerTask has returned null"))));
}
@Override
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java
index 239d36a9..7b9fd5db 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java
@@ -24,10 +24,9 @@ import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
*/
-public abstract class DmaapPublisherTask<T> extends Task {
+public abstract class DmaapPublisherTask<T, U> extends Task {
- protected abstract void publish() throws DmaapNotFoundException;
+ protected abstract U publish(U message) throws DmaapNotFoundException;
protected abstract T resolveConfiguration();
-
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java
index ff5f5893..906bd428 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java
@@ -22,6 +22,10 @@ package org.onap.dcaegen2.services.prh.tasks;
import org.onap.dcaegen2.services.config.DmaapPublisherConfiguration;
import org.onap.dcaegen2.services.prh.configuration.AppConfig;
import org.onap.dcaegen2.services.prh.configuration.Config;
+import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
+import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
+import org.onap.dcaegen2.services.service.producer.ExtendedDmaapProducerHttpClientImpl;
+import org.onap.dcaegen2.services.service.producer.ImmutableDmaapPublisherRequestDetails;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -35,11 +39,9 @@ import java.time.format.DateTimeFormatter;
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
*/
@Component
-public class DmaapPublisherTaskImpl extends DmaapPublisherTask<DmaapPublisherConfiguration> {
+public class DmaapPublisherTaskImpl extends DmaapPublisherTask<DmaapPublisherConfiguration, String> {
private static final Logger logger = LoggerFactory.getLogger(ScheduledTasks.class);
- private static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
-
private final Config prhAppConfig;
@Autowired
@@ -48,27 +50,22 @@ public class DmaapPublisherTaskImpl extends DmaapPublisherTask<DmaapPublisherCon
}
@Override
- protected void publish() {
- logger.debug("Start task DmaapPublisherTask::publish() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
-
- logger.debug("End task DmaapPublisherTask::publish() :: Execution Time - {}",
- dateTimeFormatter.format(LocalDateTime.now()));
+ protected String publish(String message) {
+ logger.trace("Method %M called with arg {}", message);
+ ExtendedDmaapProducerHttpClientImpl dmaapProducerHttpClient = new ExtendedDmaapProducerHttpClientImpl(
+ resolveConfiguration());
+ return null;
}
@Override
- public ResponseEntity execute(Object object) {
- logger.debug("Start task DmaapPublisherTask::consume() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
- publish();
- logger.debug("End task DmaapPublisherTask::consume() :: Execution Time - {}",
- dateTimeFormatter.format(LocalDateTime.now()));
- return null;
+ public Object execute(Object object) throws PrhTaskException {
+ logger.trace("Method %M called with arg {}", object);
+ return publish((String) object);
}
@Override
void initConfigs() {
-
+ logger.trace("initConfigs for DmaapPublisherTaskImpl not needed/supported");
}
@Override
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java
index c80b8a7d..84c75e08 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java
@@ -35,7 +35,6 @@ import java.time.format.DateTimeFormatter;
public class ScheduledTasks {
private static final Logger logger = LoggerFactory.getLogger(ScheduledTasks.class);
- private static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
private final Task dmaapConsumerTask;
private final Task dmaapProducerTask;
@@ -52,25 +51,21 @@ public class ScheduledTasks {
}
public void scheduleMainPrhEventTask() {
- logger.debug("Task scheduledTaskAskingDMaaPOfConsumeEvent() :: Execution Time - {}", dateTimeFormatter.format(
- LocalDateTime.now()));
+ logger.trace("Execution of task was registered");
setTaskExecutionFlow();
try {
dmaapConsumerTask.initConfigs();
dmaapConsumerTask.receiveRequest(null);
} catch (PrhTaskException e) {
logger
- .error("Task scheduledTaskAskingDMaaPOfConsumeEvent()::PrhTaskException :: Execution Time - {}:{}",
- dateTimeFormatter.format(
- LocalDateTime.now()), e);
+ .warn("Chain of tasks have been aborted, because some errors occur in prh workflow ", e);
}
}
private void setTaskExecutionFlow() {
dmaapConsumerTask.setNext(aaiProducerTask);
- aaiProducerTask.setNext(dmaapProducerTask);
- aaiConsumerTask.setNext(aaiConsumerTask);
- dmaapProducerTask.setNext(dmaapConsumerTask);
+ aaiProducerTask.setNext(aaiConsumerTask);
+ aaiConsumerTask.setNext(dmaapProducerTask);
}
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/Task.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/Task.java
index 3f521f2c..a9b10d40 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/Task.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/Task.java
@@ -37,13 +37,13 @@ public abstract class Task {
public void receiveRequest(Object body) throws PrhTaskException {
- ResponseEntity responseEntity = execute(body);
+ Object response = execute(body);
if (taskProcess != null) {
- taskProcess.receiveRequest(responseEntity);
+ taskProcess.receiveRequest(response);
}
}
- abstract ResponseEntity execute(Object object) throws PrhTaskException;
+ abstract Object execute(Object object) throws PrhTaskException;
abstract void initConfigs();
}