aboutsummaryrefslogtreecommitdiffstats
path: root/prh-app-server/src/main
diff options
context:
space:
mode:
Diffstat (limited to 'prh-app-server/src/main')
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsBootstrapConfiguration.java41
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsProperties.java21
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocator.java38
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorForAutoCommitDisabled.java94
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/MainApp.java11
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AaiHttpClientConfig.java39
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfiguration.java55
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationForAutoCommitDisabledMode.java67
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsContentParser.java3
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfig.java106
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/SwaggerConfig.java82
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleController.java22
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerForAutoCommitDisabled.java80
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java173
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiProducerTask.java3
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTask.java4
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTaskImpl.java25
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java49
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java34
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/EpochDateTimeConversion.java95
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTask.java35
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImpl.java130
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksRunnerWithCommit.java98
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksWithCommit.java203
-rw-r--r--prh-app-server/src/main/resources/application.yaml13
-rw-r--r--prh-app-server/src/main/resources/logback-spring.xml23
26 files changed, 1269 insertions, 275 deletions
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsBootstrapConfiguration.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsBootstrapConfiguration.java
index f668a581..c82c326f 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsBootstrapConfiguration.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsBootstrapConfiguration.java
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -22,26 +23,36 @@ package org.onap.dcaegen2.services.bootstrap;
import org.onap.dcaegen2.services.prh.configuration.CbsConfiguration;
+import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+
@Configuration
@EnableConfigurationProperties(CbsProperties.class)
public class CbsBootstrapConfiguration {
- private static final CbsConfiguration CBS_CONFIGURATION = new CbsConfiguration();
+
+ private static CbsConfiguration CBS_CONFIGURATION = new CbsConfiguration();
+ private static CbsConfigurationForAutoCommitDisabledMode CBS_CONFIGURATION_FOR_AUTO_COMMIT_DISABLED_MODE =
+ new CbsConfigurationForAutoCommitDisabledMode();
@Bean
- public CbsProperties cbsProperties() {
+ public CbsProperties cbsProperties()
+ {
return new CbsProperties();
}
@Bean
@ConditionalOnProperty(value = "cbs.enabled", matchIfMissing = true)
+ @Profile("!autoCommitDisabled")
public CbsPropertySourceLocator cbsPropertySourceLocator(
CbsProperties cbsProperties,
CbsConfiguration cbsConfiguration) {
+
+ System.out.println("Trying to return CbsPropertySourceLocator bean");
return new CbsPropertySourceLocator(
cbsProperties,
@@ -50,9 +61,33 @@ public class CbsBootstrapConfiguration {
new CbsClientFactoryFacade(),
cbsConfiguration);
}
+
+ @Bean
+ @ConditionalOnProperty(value = "cbs.enabled", matchIfMissing = true)
+ @Profile("autoCommitDisabled")
+ public CbsPropertySourceLocatorForAutoCommitDisabled cbsPropertySourceLocatorForAutoCommitDisabled(CbsProperties cbsProperties,
+ CbsConfigurationForAutoCommitDisabledMode cbsConfigurationforAutoCommitdisabledMode) {
+
+ System.out.println("Trying to return CbsPropertySourceLocatorForAutoCommitDisabled bean");
+
+ CbsPropertySourceLocatorForAutoCommitDisabled cbsPropertySourceLocatorACDM = new CbsPropertySourceLocatorForAutoCommitDisabled(cbsProperties,
+ new CbsJsonToPropertyMapConverter(), new CbsClientConfigurationResolver(cbsProperties),
+ new CbsClientFactoryFacade(), cbsConfigurationforAutoCommitdisabledMode);
+
+ return cbsPropertySourceLocatorACDM;
+
+ }
@Bean
+ @Profile("!autoCommitDisabled")
public CbsConfiguration cbsConfiguration() {
- return CBS_CONFIGURATION;
+ return CBS_CONFIGURATION;
+ }
+
+ @Bean
+ @Profile("autoCommitDisabled")
+ public CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode() {
+ return CBS_CONFIGURATION_FOR_AUTO_COMMIT_DISABLED_MODE;
}
+
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsProperties.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsProperties.java
index 18d4021b..5fa4cdbe 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsProperties.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsProperties.java
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -35,34 +36,14 @@ public class CbsProperties {
private Duration updatesInterval;
@NestedConfigurationProperty
private RetryProperties fetchRetries = new RetryProperties();
- private String hostname;
- private Integer port;
private String appName;
CbsClientConfiguration toCbsClientConfiguration() {
return ImmutableCbsClientConfiguration.builder()
- .hostname(hostname)
- .port(port)
.appName(appName)
.build();
}
- public String getHostname() {
- return hostname;
- }
-
- public void setHostname(String hostname) {
- this.hostname = hostname;
- }
-
- public Integer getPort() {
- return port;
- }
-
- public void setPort(Integer port) {
- this.port = port;
- }
-
public String getAppName() {
return appName;
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocator.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocator.java
index 2b5ac2e5..b4875eed 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocator.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocator.java
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2019-2021 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -28,6 +29,7 @@ import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnos
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.bootstrap.config.PropertySourceLocator;
+import org.springframework.context.annotation.Profile;
import org.springframework.core.env.Environment;
import org.springframework.core.env.MapPropertySource;
import org.springframework.core.env.PropertySource;
@@ -35,6 +37,7 @@ import reactor.util.retry.Retry;
import java.util.Map;
+@Profile("!autoCommitDisabled")
public class CbsPropertySourceLocator implements PropertySourceLocator {
private static final Logger LOGGER = LoggerFactory.getLogger(CbsPropertySourceLocator.class);
@@ -43,37 +46,40 @@ public class CbsPropertySourceLocator implements PropertySourceLocator {
private final CbsClientConfigurationResolver cbsClientConfigurationResolver;
private final CbsClientFactoryFacade cbsClientFactoryFacade;
private final CbsConfiguration cbsConfiguration;
-
+
public CbsPropertySourceLocator(CbsProperties cbsProperties,
- CbsJsonToPropertyMapConverter cbsJsonToPropertyMapConverter,
- CbsClientConfigurationResolver cbsClientConfigurationResolver,
- CbsClientFactoryFacade cbsClientFactoryFacade,
- CbsConfiguration cbsConfiguration) {
- this.cbsProperties = cbsProperties;
- this.cbsJsonToPropertyMapConverter = cbsJsonToPropertyMapConverter;
- this.cbsClientConfigurationResolver = cbsClientConfigurationResolver;
- this.cbsClientFactoryFacade = cbsClientFactoryFacade;
- this.cbsConfiguration = cbsConfiguration;
+ CbsJsonToPropertyMapConverter cbsJsonToPropertyMapConverter,
+ CbsClientConfigurationResolver cbsClientConfigurationResolver,
+ CbsClientFactoryFacade cbsClientFactoryFacade, CbsConfiguration cbsConfiguration) {
+
+ this.cbsProperties = cbsProperties;
+ this.cbsJsonToPropertyMapConverter = cbsJsonToPropertyMapConverter;
+ this.cbsClientConfigurationResolver = cbsClientConfigurationResolver;
+ this.cbsClientFactoryFacade = cbsClientFactoryFacade;
+ this.cbsConfiguration = cbsConfiguration;
}
@Override
public PropertySource<?> locate(Environment environment) {
+
CbsClientConfiguration cbsClientConfiguration = cbsClientConfigurationResolver.resolveCbsClientConfiguration();
Map<String, Object> properties = cbsClientFactoryFacade.createCbsClient(cbsClientConfiguration)
.flatMap(cbsClient -> cbsClient.get(CbsRequests.getAll(RequestDiagnosticContext.create())))
.doOnError(e -> LOGGER.warn("Failed loading configuration - retrying...", e))
- .retryWhen(Retry.
- backoff(cbsProperties.getFetchRetries().getMaxAttempts(), cbsProperties.getFetchRetries().getFirstBackoff()).
- maxBackoff(cbsProperties.getFetchRetries().getMaxBackoff()))
- .doOnNext(this::updateCbsConfig)
- .map(cbsJsonToPropertyMapConverter::convertToMap)
- .block();
+ .retryWhen(Retry
+ .backoff(cbsProperties.getFetchRetries().getMaxAttempts(),
+ cbsProperties.getFetchRetries().getFirstBackoff())
+ .maxBackoff(cbsProperties.getFetchRetries().getMaxBackoff()))
+ .doOnNext(this::updateCbsConfig).map(cbsJsonToPropertyMapConverter::convertToMap).block();
+
return new MapPropertySource("cbs", properties);
}
private void updateCbsConfig(JsonObject jsonObject) {
try {
+ LOGGER.info("Updating CBS configuration");
cbsConfiguration.parseCBSConfig(jsonObject);
+
} catch (Exception e) {
LOGGER.error("Failed parsing configuration", e);
throw e;
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorForAutoCommitDisabled.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorForAutoCommitDisabled.java
new file mode 100644
index 00000000..b7aa1f58
--- /dev/null
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorForAutoCommitDisabled.java
@@ -0,0 +1,94 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2023 Deutsche Telekom 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.bootstrap;
+
+import com.google.gson.JsonObject;
+import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.cloud.bootstrap.config.PropertySourceLocator;
+import org.springframework.context.annotation.Profile;
+import org.springframework.core.env.Environment;
+import org.springframework.core.env.MapPropertySource;
+import org.springframework.core.env.PropertySource;
+import reactor.util.retry.Retry;
+import java.util.Map;
+
+/**
+ * * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on
+ * * 24/08/23
+ * */
+
+@Profile("autoCommitDisabled")
+public class CbsPropertySourceLocatorForAutoCommitDisabled implements PropertySourceLocator {
+ private static final Logger LOGGER = LoggerFactory.getLogger(CbsPropertySourceLocatorForAutoCommitDisabled.class);
+
+ private final CbsProperties cbsProperties;
+ private final CbsJsonToPropertyMapConverter cbsJsonToPropertyMapConverter;
+ private final CbsClientConfigurationResolver cbsClientConfigurationResolver;
+ private final CbsClientFactoryFacade cbsClientFactoryFacade;
+ private final CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode;
+
+ public CbsPropertySourceLocatorForAutoCommitDisabled(CbsProperties cbsProperties,
+ CbsJsonToPropertyMapConverter cbsJsonToPropertyMapConverter,
+ CbsClientConfigurationResolver cbsClientConfigurationResolver,
+ CbsClientFactoryFacade cbsClientFactoryFacade, CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode) {
+
+ this.cbsProperties = cbsProperties;
+ this.cbsJsonToPropertyMapConverter = cbsJsonToPropertyMapConverter;
+ this.cbsClientConfigurationResolver = cbsClientConfigurationResolver;
+ this.cbsClientFactoryFacade = cbsClientFactoryFacade;
+ this.cbsConfigurationForAutoCommitDisabledMode = cbsConfigurationForAutoCommitDisabledMode;
+
+ }
+
+ @Override
+ public PropertySource<?> locate(Environment environment) {
+
+ CbsClientConfiguration cbsClientConfiguration = cbsClientConfigurationResolver.resolveCbsClientConfiguration();
+ Map<String, Object> properties = cbsClientFactoryFacade.createCbsClient(cbsClientConfiguration)
+ .flatMap(cbsClient -> cbsClient.get(CbsRequests.getAll(RequestDiagnosticContext.create())))
+ .doOnError(e -> LOGGER.warn("Failed loading configuration - retrying...", e))
+ .retryWhen(Retry
+ .backoff(cbsProperties.getFetchRetries().getMaxAttempts(),
+ cbsProperties.getFetchRetries().getFirstBackoff())
+ .maxBackoff(cbsProperties.getFetchRetries().getMaxBackoff()))
+ .doOnNext(this::updateCbsConfig)
+ .map(cbsJsonToPropertyMapConverter::convertToMap).block();
+
+ return new MapPropertySource("cbs", properties);
+ }
+
+ private void updateCbsConfig(JsonObject jsonObject) {
+ try {
+ LOGGER.info("Updating CBS configuration");
+ cbsConfigurationForAutoCommitDisabledMode.parseCBSConfig(jsonObject);
+
+ } catch (Exception e) {
+ LOGGER.error("Failed parsing configuration", e);
+ throw e;
+ }
+ }
+
+}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/MainApp.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/MainApp.java
index 1d2a65d3..5a986517 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/MainApp.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/MainApp.java
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -22,7 +23,6 @@ package org.onap.dcaegen2.services.prh;
import java.util.Map;
import java.util.UUID;
-
import org.slf4j.MDC;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -32,19 +32,22 @@ import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler;
-
import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.INVOCATION_ID;
/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
+ * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on
+ * 3/23/18
*/
-@SpringBootApplication(exclude = {JacksonAutoConfiguration.class})
+@SpringBootApplication(exclude = { JacksonAutoConfiguration.class })
@EnableScheduling
@EnableConfigurationProperties
public class MainApp {
+
public static void main(String[] args) {
+
SpringApplication.run(MainApp.class, args);
+
}
@Bean
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AaiHttpClientConfig.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AaiHttpClientConfig.java
index 39369329..cd0d8d2a 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AaiHttpClientConfig.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AaiHttpClientConfig.java
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -21,6 +22,7 @@
package org.onap.dcaegen2.services.prh.configuration;
import java.nio.charset.StandardCharsets;
+
import java.util.function.BiFunction;
import org.onap.dcaegen2.services.prh.adapter.aai.api.AaiHttpClient;
import org.onap.dcaegen2.services.prh.adapter.aai.api.AaiPnfResultModel;
@@ -39,47 +41,48 @@ import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
@Configuration
public class AaiHttpClientConfig {
@Autowired
- private CbsConfiguration cbsConfiguration;
+ private Config config;
@Bean
public AaiHttpClient<ConsumerDmaapModel, HttpResponse> getPatchClientFactory() {
return createLazyConfigClient(
- (config, client) -> new AaiHttpPatchClient(config, new AaiJsonBodyBuilderImpl(), client));
+ (config, client) -> new AaiHttpPatchClient(config, new AaiJsonBodyBuilderImpl(), client));
}
@Bean
public AaiHttpClient<AaiServiceInstanceQueryModel, AaiServiceInstanceResultModel> getServiceInstanceClient() {
return createLazyConfigClient(
- (config, client) -> new AaiGetServiceInstanceClient(config, client)
- .map(httpResponse -> {
+ (config, client) -> new AaiGetServiceInstanceClient(config, client).map(httpResponse -> {
httpResponse.throwIfUnsuccessful();
- return httpResponse.bodyAsJson(StandardCharsets.UTF_8,
- PrhModelAwareGsonBuilder.createGson(), AaiServiceInstanceResultModel.class);
+ return httpResponse.bodyAsJson(StandardCharsets.UTF_8, PrhModelAwareGsonBuilder.createGson(),
+ AaiServiceInstanceResultModel.class);
}));
}
@Bean
public AaiHttpClient<ConsumerDmaapModel, AaiPnfResultModel> getGetClient() {
- return createLazyConfigClient(
- (config, client) -> new AaiHttpGetClient(config, client)
- .map(httpResponse -> {
- httpResponse.throwIfUnsuccessful();
- return httpResponse.bodyAsJson(StandardCharsets.UTF_8,
- PrhModelAwareGsonBuilder.createGson(), AaiPnfResultModel.class);
- }));
+
+
+
+ return createLazyConfigClient((config, client) -> new AaiHttpGetClient(config, client).map(httpResponse -> {
+ httpResponse.throwIfUnsuccessful();
+ return httpResponse.bodyAsJson(StandardCharsets.UTF_8, PrhModelAwareGsonBuilder.createGson(),
+ AaiPnfResultModel.class);
+ }));
}
private <T, U> AaiHttpClient<T, U> createLazyConfigClient(
- final BiFunction<AaiClientConfiguration, RxHttpClient, AaiHttpClient<T, U>> factoryMethod) {
+ final BiFunction<AaiClientConfiguration, RxHttpClient, AaiHttpClient<T, U>> factoryMethod) {
+// System.out.println("pnf url in AAIClientConfiguration is: " + config.getAaiClientConfiguration().pnfUrl());
+// System.out.println("base url in AAIClientConfiguration is: " + config.getAaiClientConfiguration().baseUrl());
+ return x -> factoryMethod.apply(config.getAaiClientConfiguration(),
+ new AaiHttpClientFactory(config.getAaiClientConfiguration()).build()).getAaiResponse(x);
- return x -> factoryMethod.apply(
- cbsConfiguration.getAaiClientConfiguration(),
- new AaiHttpClientFactory(cbsConfiguration.getAaiClientConfiguration()).build()
- ).getAaiResponse(x);
}
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfiguration.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfiguration.java
index 8373018d..22763e8b 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfiguration.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfiguration.java
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -30,35 +31,42 @@ import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRo
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+@Profile("!autoCommitDisabled")
public class CbsConfiguration implements Config {
private static final Logger LOGGER = LoggerFactory.getLogger(CbsConfiguration.class);
- private static final String CBS_CONFIG_MISSING = "CBS config missing";
- private AaiClientConfiguration aaiClientCBSConfiguration;
- private MessageRouterPublisher messageRouterPublisher;
- private MessageRouterSubscriber messageRouterSubscriber;
- private MessageRouterPublishRequest messageRouterCBSPublishRequest;
- private MessageRouterSubscribeRequest messageRouterCBSSubscribeRequest;
- private MessageRouterPublishRequest messageRouterCBSUpdatePublishRequest;
-
-
+ protected static final String CBS_CONFIG_MISSING = "CBS config missing";
+ protected AaiClientConfiguration aaiClientCBSConfiguration;
+ protected MessageRouterPublisher messageRouterPublisher;
+ protected MessageRouterSubscriber messageRouterSubscriber;
+ protected MessageRouterPublishRequest messageRouterCBSPublishRequest;
+ protected MessageRouterSubscribeRequest messageRouterCBSSubscribeRequest;
+ protected MessageRouterPublishRequest messageRouterCBSUpdatePublishRequest;
+
public void parseCBSConfig(JsonObject jsonObject) {
+
LOGGER.info("Received application configuration: {}", jsonObject);
- CbsContentParser consulConfigurationParser = new CbsContentParser(jsonObject);
-
+ CbsContentParser consulConfigurationParser = new CbsContentParser(jsonObject);
aaiClientCBSConfiguration = consulConfigurationParser.getAaiClientConfig();
- messageRouterPublisher = DmaapClientFactory.createMessageRouterPublisher(
- consulConfigurationParser.getMessageRouterPublisherConfig());
+ messageRouterPublisher = DmaapClientFactory
+ .createMessageRouterPublisher(consulConfigurationParser.getMessageRouterPublisherConfig());
messageRouterCBSPublishRequest = consulConfigurationParser.getMessageRouterPublishRequest();
messageRouterCBSUpdatePublishRequest = consulConfigurationParser.getMessageRouterUpdatePublishRequest();
- messageRouterSubscriber = DmaapClientFactory.createMessageRouterSubscriber(
- consulConfigurationParser.getMessageRouterSubscriberConfig());
+ messageRouterSubscriber = DmaapClientFactory
+ .createMessageRouterSubscriber(consulConfigurationParser.getMessageRouterSubscriberConfig());
+ String prevTopicUrl = null;
+ if(messageRouterCBSSubscribeRequest != null) {
+ prevTopicUrl = messageRouterCBSSubscribeRequest.sourceDefinition().topicUrl();
+ }
messageRouterCBSSubscribeRequest = consulConfigurationParser.getMessageRouterSubscribeRequest();
- }
-
+ if(!messageRouterCBSSubscribeRequest.sourceDefinition().topicUrl().equals(prevTopicUrl)) {
+ messageRouterSubscriber.close();
+ }
+ }
@Override
public MessageRouterPublisher getMessageRouterPublisher() {
@@ -72,21 +80,26 @@ public class CbsConfiguration implements Config {
@Override
public MessageRouterPublishRequest getMessageRouterPublishRequest() {
- return Optional.ofNullable(messageRouterCBSPublishRequest).orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING));
+ return Optional.ofNullable(messageRouterCBSPublishRequest)
+ .orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING));
}
@Override
public MessageRouterPublishRequest getMessageRouterUpdatePublishRequest() {
- return Optional.ofNullable(messageRouterCBSUpdatePublishRequest).orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING));
+ return Optional.ofNullable(messageRouterCBSUpdatePublishRequest)
+ .orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING));
}
@Override
public AaiClientConfiguration getAaiClientConfiguration() {
- return Optional.ofNullable(aaiClientCBSConfiguration).orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING));
+ return Optional.ofNullable(aaiClientCBSConfiguration)
+ .orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING));
}
@Override
public MessageRouterSubscribeRequest getMessageRouterSubscribeRequest() {
- return Optional.ofNullable(messageRouterCBSSubscribeRequest).orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING));
+ return Optional.ofNullable(messageRouterCBSSubscribeRequest)
+ .orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING));
}
+
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationForAutoCommitDisabledMode.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationForAutoCommitDisabledMode.java
new file mode 100644
index 00000000..b20cbad6
--- /dev/null
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationForAutoCommitDisabledMode.java
@@ -0,0 +1,67 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2023 Deutsche Telekom 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.configuration;
+
+import java.util.Optional;
+import org.onap.dcaegen2.services.prh.adapter.kafka.ImmutableKafkaConfiguration;
+import org.onap.dcaegen2.services.prh.adapter.kafka.KafkaConfiguration;
+import org.springframework.context.annotation.Profile;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+
+/**
+ * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on
+ * 24/08/23
+ */
+@Profile("autoCommitDisabled")
+public class CbsConfigurationForAutoCommitDisabledMode extends CbsConfiguration {
+
+ protected KafkaConfiguration kafkaConfiguration;
+
+ @Override
+ public void parseCBSConfig(JsonObject jsonObject) {
+
+ super.parseCBSConfig(jsonObject);
+ JsonObject jsonObjectforAutoCommitDisabled = jsonObject.getAsJsonObject("config");
+ JsonElement jsonObjectOfKafkaConfigurations = jsonObjectforAutoCommitDisabled.get("kafka-configurations");
+
+ kafkaConfiguration = new ImmutableKafkaConfiguration.Builder()
+ .kafkaBoostrapServerConfig(
+ ((JsonObject) jsonObjectOfKafkaConfigurations).get("kafkaBoostrapServerConfig").getAsString())
+ .groupIdConfig(((JsonObject) jsonObjectOfKafkaConfigurations).get("groupIdConfig").getAsString())
+ .kafkaSaslMechanism(
+ ((JsonObject) jsonObjectOfKafkaConfigurations).get("kafkaSaslMechanism").getAsString())
+ .kafkaSecurityProtocol(
+ ((JsonObject) jsonObjectOfKafkaConfigurations).get("kafkaSecurityProtocol").getAsString())
+ .kafkaJaasConfig(System.getenv("JAAS_CONFIG"))
+ .build();
+
+ }
+
+ public KafkaConfiguration getKafkaConfig() {
+ return Optional.ofNullable(kafkaConfiguration).orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING));
+ }
+
+ public void setKafkaConfiguration(KafkaConfiguration kafkaConfiguration) {
+ this.kafkaConfiguration = kafkaConfiguration;
+ }
+
+}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsContentParser.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsContentParser.java
index ed935501..e1200119 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsContentParser.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsContentParser.java
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -139,4 +140,4 @@ class CbsContentParser {
.timeout(Duration.ofMillis(jsonObject.get("dmaap.dmaapConsumerConfiguration.timeoutMs").getAsLong()))
.build();
}
-} \ No newline at end of file
+}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfig.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfig.java
new file mode 100644
index 00000000..baaf3b16
--- /dev/null
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfig.java
@@ -0,0 +1,106 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2023 Deutsche Telekom 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.configuration;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.springframework.context.annotation.Profile;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
+
+ /**
+ * * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on
+ * * 24/08/23
+ * */
+
+@Profile("autoCommitDisabled")
+@EnableKafka
+@Configuration
+public class KafkaConfig {
+
+ CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode;
+
+ public String kafkaBoostrapServerConfig;
+ public String groupIdConfig;
+ public String kafkaSecurityProtocol;
+ public String kafkaSaslMechanism;
+ public String kafkaUsername;
+ public String kafkaPassword;
+ public String kafkaJaasConfigName;
+ public String kafkaLoginModuleClassConfig;
+ public String kafkaJaasConfig;
+
+ public final String DEFAULT_KAFKA_SECURITY_PROTOCOL = "SASL_PLAINTEXT";
+ public final String DEFAULT_KAFKA_SASL_MECHANISM = "SCRAM-SHA-512";
+
+ public KafkaConfig() {
+
+ }
+
+ @Bean
+ public ConsumerFactory<String, String> consumerFactory(CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode) {
+ this.cbsConfigurationForAutoCommitDisabledMode = cbsConfigurationForAutoCommitDisabledMode;
+ kafkaBoostrapServerConfig = cbsConfigurationForAutoCommitDisabledMode.getKafkaConfig()
+ .kafkaBoostrapServerConfig();
+ groupIdConfig = cbsConfigurationForAutoCommitDisabledMode.getKafkaConfig().groupIdConfig();
+ kafkaSecurityProtocol = cbsConfigurationForAutoCommitDisabledMode.getKafkaConfig().kafkaSecurityProtocol();
+ kafkaSaslMechanism = cbsConfigurationForAutoCommitDisabledMode.getKafkaConfig().kafkaSaslMechanism();
+ kafkaJaasConfig = cbsConfigurationForAutoCommitDisabledMode.getKafkaConfig().kafkaJaasConfig();
+
+ Map<String, Object> config = new HashMap<>();
+ config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBoostrapServerConfig);
+
+ config.put(ConsumerConfig.GROUP_ID_CONFIG, groupIdConfig);
+ config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+ "org.apache.kafka.common.serialization.StringDeserializer");
+ config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+ "org.apache.kafka.common.serialization.StringDeserializer");
+ config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+
+ if (kafkaSecurityProtocol == null)
+ kafkaSecurityProtocol = DEFAULT_KAFKA_SECURITY_PROTOCOL;
+ config.put("security.protocol", kafkaSecurityProtocol);
+ if (kafkaSaslMechanism == null)
+ kafkaSaslMechanism = DEFAULT_KAFKA_SASL_MECHANISM;
+ config.put("sasl.mechanism", kafkaSaslMechanism);
+
+ config.put("sasl.jaas.config", kafkaJaasConfig);
+
+ return new DefaultKafkaConsumerFactory<>(config);
+
+ }
+
+ @Bean
+ public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
+ CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode) {
+ ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
+ factory.setConsumerFactory(consumerFactory(cbsConfigurationForAutoCommitDisabledMode));
+ factory.setBatchListener(true);
+ factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
+ return factory;
+ }
+
+}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/SwaggerConfig.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/SwaggerConfig.java
deleted file mode 100644
index 4039f698..00000000
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/SwaggerConfig.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * PNF-REGISTRATION-HANDLER
- * ================================================================================
- * Copyright (C) 2018-2020 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.configuration;
-
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Profile;
-import org.springframework.web.servlet.config.annotation.ResourceHandlerRegistry;
-import org.springframework.web.servlet.config.annotation.WebMvcConfigurationSupport;
-import springfox.documentation.builders.ApiInfoBuilder;
-import springfox.documentation.builders.PathSelectors;
-import springfox.documentation.builders.RequestHandlerSelectors;
-import springfox.documentation.service.ApiInfo;
-import springfox.documentation.spi.DocumentationType;
-import springfox.documentation.spring.web.plugins.Docket;
-import springfox.documentation.swagger2.annotations.EnableSwagger2;
-
-
-@Configuration
-@Profile("prod")
-public class SwaggerConfig extends WebMvcConfigurationSupport {
-
- private static final String PACKAGE_PATH = "org.onap.dcaegen2.services.prh";
- private static final String API_TITLE = "PRH app server";
- private static final String DESCRIPTION = "This page lists all the rest apis for PRH app server.";
- private static final String VERSION = "1.0";
- private static final String RESOURCES_PATH = "classpath:/META-INF/resources/";
- private static final String WEBJARS_PATH = RESOURCES_PATH + "webjars/";
- private static final String SWAGGER_UI = "swagger-ui.html";
- private static final String WEBJARS = "/webjars/**";
-
- /**
- * Swagger configuration function for hosting it next to spring http website.
- *
- * @return Docket
- */
- @Bean
- public Docket api() {
- return new Docket(DocumentationType.SWAGGER_2)
- .apiInfo(apiInfo())
- .select()
- .apis(RequestHandlerSelectors.basePackage(PACKAGE_PATH))
- .paths(PathSelectors.any())
- .build();
- }
-
- private ApiInfo apiInfo() {
- return new ApiInfoBuilder()
- .title(API_TITLE)
- .description(DESCRIPTION)
- .version(VERSION)
- .build();
- }
-
-
- @Override
- protected void addResourceHandlers(ResourceHandlerRegistry registry) {
- registry.addResourceHandler(SWAGGER_UI)
- .addResourceLocations(RESOURCES_PATH);
-
- registry.addResourceHandler(WEBJARS)
- .addResourceLocations(WEBJARS_PATH);
- }
-}
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 a0aa17e3..aafcd81a 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
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -26,6 +27,7 @@ import org.onap.dcaegen2.services.prh.tasks.ScheduledTasksRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Profile;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.RequestMapping;
@@ -34,17 +36,19 @@ import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/5/18
+ * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on
+ * 4/5/18
*/
@RestController
-@Api(value = "ScheduleController", description = "Schedule Controller")
+@Api(value = "ScheduleController")
+@Profile("!autoCommitDisabled")
public class ScheduleController {
private static final Logger LOGGER = LoggerFactory.getLogger(ScheduleController.class);
+ private ScheduledTasksRunner scheduledTasksRunner;
- private final ScheduledTasksRunner scheduledTasksRunner;
- @Autowired
+ @Autowired(required = false)
public ScheduleController(ScheduledTasksRunner scheduledTasksRunner) {
this.scheduledTasksRunner = scheduledTasksRunner;
}
@@ -52,7 +56,7 @@ public class ScheduleController {
@RequestMapping(value = "start", method = RequestMethod.GET)
@ApiOperation(value = "Start scheduling worker request")
public Mono<ResponseEntity<String>> startTasks() {
- LOGGER.trace("Receiving start scheduling worker request");
+ LOGGER.trace("Receiving start scheduling worker request with Comit SchedulerController");
return Mono.fromSupplier(scheduledTasksRunner::tryToStartTask).map(this::createStartTaskResponse);
}
@@ -61,10 +65,10 @@ public class ScheduleController {
public Mono<ResponseEntity<String>> stopTask() {
LOGGER.trace("Receiving stop scheduling worker request");
return Mono.defer(() -> {
- scheduledTasksRunner.cancelTasks();
- return Mono.just(new ResponseEntity<>("PRH Service has been stopped!", HttpStatus.OK));
- }
- );
+ scheduledTasksRunner.closeKafkaPublisherSubscriber();
+ scheduledTasksRunner.cancelTasks();
+ return Mono.just(new ResponseEntity<>("PRH Service has been stopped!", HttpStatus.OK));
+ });
}
private ResponseEntity<String> createStartTaskResponse(boolean wasScheduled) {
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerForAutoCommitDisabled.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerForAutoCommitDisabled.java
new file mode 100644
index 00000000..64e3a469
--- /dev/null
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerForAutoCommitDisabled.java
@@ -0,0 +1,80 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2023 Deutsche Telekom 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.controllers;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import org.onap.dcaegen2.services.prh.tasks.commit.ScheduledTasksRunnerWithCommit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Profile;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
+
+/**
+ * * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on
+ * * 24/08/23
+ * */
+@RestController
+@Api(value = "ScheduleController")
+@Profile("autoCommitDisabled")
+public class ScheduleControllerForAutoCommitDisabled {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ScheduleControllerForAutoCommitDisabled.class);
+
+ private ScheduledTasksRunnerWithCommit scheduledTasksRunnerWithCommit;
+
+ @Autowired(required = false)
+ public ScheduleControllerForAutoCommitDisabled(ScheduledTasksRunnerWithCommit scheduledTasksRunnerWithCommit) {
+ this.scheduledTasksRunnerWithCommit = scheduledTasksRunnerWithCommit;
+ }
+
+ @RequestMapping(value = "start", method = RequestMethod.GET)
+ @ApiOperation(value = "Start scheduling worker request")
+ public Mono<ResponseEntity<String>> startTasks() {
+ LOGGER.trace("Receiving start scheduling worker request with Comit SchedulerController");
+ return Mono.fromSupplier(scheduledTasksRunnerWithCommit::tryToStartTaskWithCommit)
+ .map(this::createStartTaskResponse);
+ }
+
+ @RequestMapping(value = "stopPrh", method = RequestMethod.GET)
+ @ApiOperation(value = "Receiving stop scheduling worker request")
+ public Mono<ResponseEntity<String>> stopTask() {
+ LOGGER.trace("Receiving stop scheduling worker request");
+ return Mono.defer(() -> {
+ scheduledTasksRunnerWithCommit.cancelTasks();
+ return Mono.just(new ResponseEntity<>("PRH Service has been stopped!", HttpStatus.OK));
+ });
+ }
+
+ private ResponseEntity<String> createStartTaskResponse(boolean wasScheduled) {
+ if (wasScheduled) {
+ return new ResponseEntity<>("PRH Service has been started!", HttpStatus.CREATED);
+ } else {
+ return new ResponseEntity<>("PRH Service is already running!", HttpStatus.NOT_ACCEPTABLE);
+ }
+ }
+}
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
index f98e952f..25c380fb 100644
--- 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
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -17,43 +18,48 @@
* limitations under the License.
* ============LICENSE_END=========================================================
*/
-
package org.onap.dcaegen2.services.prh.service;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.ADDITIONAL_FIELDS;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.COMMON_EVENT_HEADER;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.COMMON_FORMAT_FOR_JSON_OBJECT;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.COMMON_FORMAT_FOR_STRING;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.CORRELATION_ID;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.EQUIP_MODEL;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.EQUIP_TYPE;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.EQUIP_VENDOR;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.EVENT;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.NF_ROLE;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.OAM_IPV_4_ADDRESS;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.OAM_IPV_6_ADDRESS;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.PNF_REGISTRATION_FIELDS;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.SERIAL_NUMBER;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.SOURCE_NAME;
-import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.SW_VERSION;
-
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import io.vavr.collection.List;
-import java.util.Optional;
-import java.util.stream.StreamSupport;
import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.adapter.aai.api.ImmutableConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.boot.configurationprocessor.json.JSONArray;
+import org.springframework.boot.configurationprocessor.json.JSONException;
+import org.springframework.boot.configurationprocessor.json.JSONObject;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import java.util.ArrayList;
+import java.util.Optional;
+import java.util.stream.StreamSupport;
+
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.SOURCE_NAME;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.OAM_IPV_4_ADDRESS;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.OAM_IPV_6_ADDRESS;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.SERIAL_NUMBER;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.EQUIP_VENDOR;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.EQUIP_MODEL;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.EQUIP_TYPE;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.NF_ROLE;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.SW_VERSION;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.ADDITIONAL_FIELDS;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.COMMON_FORMAT_FOR_STRING;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.EVENT;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.COMMON_EVENT_HEADER;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.PNF_REGISTRATION_FIELDS;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.CORRELATION_ID;
+import static org.onap.dcaegen2.services.prh.service.PnfRegistrationFields.COMMON_FORMAT_FOR_JSON_OBJECT;
+
+
+
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/8/18
@@ -74,6 +80,8 @@ public class DmaapConsumerJsonParser {
private String pnfSwVersionOptionalField;
private JsonObject pnfAdditionalFields;
+ private String sourceName;
+
/**
* Extract info from string and create @see {@link ConsumerDmaapModel}.
*
@@ -84,6 +92,11 @@ public class DmaapConsumerJsonParser {
return monoMessage.flatMapMany(msgRouterResponse -> getConsumerDmaapModelFromJsonArray(msgRouterResponse.items()));
}
+ public JSONObject getJsonObjectKafka(String jsonStr) throws JSONException {
+ return new JSONObject(jsonStr);
+ }
+
+
private Flux<ConsumerDmaapModel> getConsumerDmaapModelFromJsonArray(List<JsonElement> items) {
LOGGER.debug("DmaapConsumerJsonParser input for parsing: {}", items);
@@ -97,25 +110,59 @@ public class DmaapConsumerJsonParser {
.orElseGet(JsonObject::new)))));
}
+ /**
+ * Extract info from string and create @see {@link ConsumerDmaapModel}.
+ *
+ * @param monoMessage - results from Kafka
+ * @return reactive DMaaPModel
+ *
+ */
+ /**
+ * @author <a href="mailto:shilpa.urade@t-systems.com">Shilpa Urade</a> on 13/3/23
+ */
+
+ public Flux<ConsumerDmaapModel> getConsumerDmaapModelFromKafkaConsumerRecord(java.util.List<String> items)
+ {
+ LOGGER.info("DmaapConsumerJsonParser input for parsing: {} with commit", items);
+ if (items.size() == 0) {
+ LOGGER.info("Nothing to consume from Kafka");
+ return Flux.empty();
+ }
+ return create(
+ Flux.defer(() -> Flux.fromStream(StreamSupport.stream(items.spliterator(), false)
+ .map(jsonObjectFromString -> getJsonObjectFromString(jsonObjectFromString)
+ .orElseGet(JsonObject::new)))));
+ }
+
+ Optional<JsonObject> getJsonObjectFromString(String element) {
+ return Optional.ofNullable(JsonParser.parseString(element).getAsJsonObject());
+ }
+
+ public String getSourceName() {
+ return sourceName;
+ }
+
Optional<JsonObject> getJsonObjectFromAnArray(JsonElement element) {
JsonParser jsonParser = new JsonParser();
return element.isJsonPrimitive() ? Optional.of(jsonParser.parse(element.getAsString()).getAsJsonObject())
- : Optional.of(jsonParser.parse(element.toString()).getAsJsonObject());
+ : Optional.of(jsonParser.parse(element.toString()).getAsJsonObject());
+ }
+
+ Optional<JsonObject> getJsonObjectFromKafkaRecords(String element) {
+ return Optional.ofNullable(new JsonObject().getAsJsonObject(element));
}
+
private Flux<ConsumerDmaapModel> create(Flux<JsonObject> jsonObject) {
- return jsonObject.flatMap(monoJsonP ->
- !containsHeader(monoJsonP) ? logErrorAndReturnMonoEmpty("Incorrect JsonObject - missing header")
- : transform(monoJsonP))
- .onErrorResume(exception -> exception instanceof DmaapNotFoundException, e -> Mono.empty());
+ return jsonObject.flatMap(monoJsonP -> !containsHeader(monoJsonP) ? logErrorAndReturnMonoEmpty("Incorrect JsonObject - missing header")
+ : transform(monoJsonP));
}
private Mono<ConsumerDmaapModel> transform(JsonObject responseFromDmaap) {
JsonObject commonEventHeader = responseFromDmaap.getAsJsonObject(EVENT)
- .getAsJsonObject(COMMON_EVENT_HEADER);
+ .getAsJsonObject(COMMON_EVENT_HEADER);
JsonObject pnfRegistrationFields = responseFromDmaap.getAsJsonObject(EVENT)
- .getAsJsonObject(PNF_REGISTRATION_FIELDS);
-
+ .getAsJsonObject(PNF_REGISTRATION_FIELDS);
this.pnfSourceName = getValueFromJson(commonEventHeader, SOURCE_NAME);
this.pnfNfRoleOptionalField = getValueFromJson(commonEventHeader, NF_ROLE);
this.pnfOamIpv4Address = getValueFromJson(pnfRegistrationFields, OAM_IPV_4_ADDRESS);
@@ -126,21 +173,20 @@ public class DmaapConsumerJsonParser {
this.pnfEquipTypeOptionalField = getValueFromJson(pnfRegistrationFields, EQUIP_TYPE);
this.pnfSwVersionOptionalField = getValueFromJson(pnfRegistrationFields, SW_VERSION);
this.pnfAdditionalFields = pnfRegistrationFields.getAsJsonObject(ADDITIONAL_FIELDS);
-
return (StringUtils.isEmpty(pnfSourceName))
- ? logErrorAndReturnMonoEmpty("Incorrect json, consumerDmaapModel can not be created: "
- + printMessage()) :
- Mono.just(ImmutableConsumerDmaapModel.builder()
- .correlationId(pnfSourceName)
- .ipv4(pnfOamIpv4Address)
- .ipv6(pnfOamIpv6Address)
- .serialNumber(pnfSerialNumberOptionalField)
- .equipVendor(pnfEquipVendorOptionalField)
- .equipModel(pnfEquipModelOptionalField)
- .equipType(pnfEquipTypeOptionalField)
- .nfRole(pnfNfRoleOptionalField)
- .swVersion(pnfSwVersionOptionalField)
- .additionalFields(pnfAdditionalFields).build());
+ ? logErrorAndReturnMonoEmpty("Incorrect json, consumerDmaapModel can not be created: "
+ + printMessage()) :
+ Mono.just(ImmutableConsumerDmaapModel.builder()
+ .correlationId(pnfSourceName)
+ .ipv4(pnfOamIpv4Address)
+ .ipv6(pnfOamIpv6Address)
+ .serialNumber(pnfSerialNumberOptionalField)
+ .equipVendor(pnfEquipVendorOptionalField)
+ .equipModel(pnfEquipModelOptionalField)
+ .equipType(pnfEquipTypeOptionalField)
+ .nfRole(pnfNfRoleOptionalField)
+ .swVersion(pnfSwVersionOptionalField)
+ .additionalFields(pnfAdditionalFields).build());
}
private String getValueFromJson(JsonObject jsonObject, String jsonKey) {
@@ -148,30 +194,39 @@ public class DmaapConsumerJsonParser {
}
private boolean containsHeader(JsonObject jsonObject) {
- return jsonObject.has(EVENT) && jsonObject.getAsJsonObject(EVENT).has(PNF_REGISTRATION_FIELDS);
+ try {
+ return jsonObject.has(EVENT) && jsonObject.getAsJsonObject(EVENT).has(PNF_REGISTRATION_FIELDS);
+ }catch(Exception e){
+ LOGGER.info("Fetching an error in containsHeader method {}",e.getMessage());
+ }
+ return false;
}
private String printMessage() {
return String.format("%n{"
- + "\"" + CORRELATION_ID + COMMON_FORMAT_FOR_STRING + ","
- + "\"" + OAM_IPV_4_ADDRESS + COMMON_FORMAT_FOR_STRING + ","
- + "\"" + OAM_IPV_6_ADDRESS + COMMON_FORMAT_FOR_STRING + ","
- + "\"" + SERIAL_NUMBER + COMMON_FORMAT_FOR_STRING + ","
- + "\"" + EQUIP_VENDOR + COMMON_FORMAT_FOR_STRING + ","
- + "\"" + EQUIP_MODEL + COMMON_FORMAT_FOR_STRING + ","
- + "\"" + EQUIP_TYPE + COMMON_FORMAT_FOR_STRING + ","
- + "\"" + NF_ROLE + COMMON_FORMAT_FOR_STRING + ","
- + "\"" + SW_VERSION + COMMON_FORMAT_FOR_STRING + ","
- + "\"" + ADDITIONAL_FIELDS + COMMON_FORMAT_FOR_JSON_OBJECT
- + "%n}", this.pnfSourceName, this.pnfOamIpv4Address, this.pnfOamIpv6Address,
- this.pnfSerialNumberOptionalField, this.pnfEquipVendorOptionalField,
- this.pnfEquipModelOptionalField, this.pnfEquipTypeOptionalField,
- this.pnfNfRoleOptionalField, this.pnfSwVersionOptionalField, this.pnfAdditionalFields
+ + "\"" + CORRELATION_ID + COMMON_FORMAT_FOR_STRING + ","
+ + "\"" + OAM_IPV_4_ADDRESS + COMMON_FORMAT_FOR_STRING + ","
+ + "\"" + OAM_IPV_6_ADDRESS + COMMON_FORMAT_FOR_STRING + ","
+ + "\"" + SERIAL_NUMBER + COMMON_FORMAT_FOR_STRING + ","
+ + "\"" + EQUIP_VENDOR + COMMON_FORMAT_FOR_STRING + ","
+ + "\"" + EQUIP_MODEL + COMMON_FORMAT_FOR_STRING + ","
+ + "\"" + EQUIP_TYPE + COMMON_FORMAT_FOR_STRING + ","
+ + "\"" + NF_ROLE + COMMON_FORMAT_FOR_STRING + ","
+ + "\"" + SW_VERSION + COMMON_FORMAT_FOR_STRING + ","
+ + "\"" + ADDITIONAL_FIELDS + COMMON_FORMAT_FOR_JSON_OBJECT
+ + "%n}", this.pnfSourceName, this.pnfOamIpv4Address, this.pnfOamIpv6Address,
+ this.pnfSerialNumberOptionalField, this.pnfEquipVendorOptionalField,
+ this.pnfEquipModelOptionalField, this.pnfEquipTypeOptionalField,
+ this.pnfNfRoleOptionalField, this.pnfSwVersionOptionalField, this.pnfAdditionalFields
);
}
private <T> Mono<T> logErrorAndReturnMonoEmpty(String messageForLogger) {
- LOGGER.warn(messageForLogger);
+ LOGGER.info(messageForLogger);
return Mono.empty();
}
+
+ public JSONArray getJsonArray(String value) throws JSONException {
+ return new JSONArray(value);
+ }
}
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 35eb948b..ce8059b2 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
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -25,6 +26,6 @@ import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
import reactor.core.publisher.Mono;
@FunctionalInterface
-interface AaiProducerTask {
+public interface AaiProducerTask {
Mono<ConsumerDmaapModel> execute(ConsumerDmaapModel consumerDmaapModel) throws PrhTaskException;
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTask.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTask.java
index 11ff369a..5f86010a 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTask.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTask.java
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -23,7 +24,8 @@ package org.onap.dcaegen2.services.prh.tasks;
import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel;
import reactor.core.publisher.Mono;
-@FunctionalInterface
+
public interface AaiQueryTask {
Mono<Boolean> execute(final ConsumerDmaapModel aaiModel);
+ Mono<ConsumerDmaapModel> findPnfinAAI(final ConsumerDmaapModel model);
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTaskImpl.java
index 3db4887a..73131926 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTaskImpl.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTaskImpl.java
@@ -3,10 +3,10 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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
*
@@ -35,6 +35,8 @@ import org.onap.dcaegen2.services.prh.model.RelationshipDict;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
@Component
public class AaiQueryTaskImpl implements AaiQueryTask {
@@ -44,6 +46,7 @@ public class AaiQueryTaskImpl implements AaiQueryTask {
static final String SERVICE_TYPE = "service-subscription.service-type";
static final String SERVICE_INSTANCE_ID = "service-instance.service-instance-id";
+ private static final Logger LOGGER = LoggerFactory.getLogger(AaiQueryTaskImpl.class);
private final AaiHttpClient<ConsumerDmaapModel, AaiPnfResultModel> getPnfModelClient;
private final AaiHttpClient<AaiServiceInstanceQueryModel, AaiServiceInstanceResultModel> getServiceClient;
@@ -55,8 +58,11 @@ public class AaiQueryTaskImpl implements AaiQueryTask {
this.getServiceClient = getServiceClient;
}
+
+
@Override
public Mono<Boolean> execute(ConsumerDmaapModel aaiModel) {
+
return getPnfModelClient
.getAaiResponse(aaiModel)
.flatMap(this::checkIfPnfHasRelationToService)
@@ -65,7 +71,20 @@ public class AaiQueryTaskImpl implements AaiQueryTask {
.defaultIfEmpty(false);
}
+
+ // Added by DTAG, March 2023
+ @Override
+ public Mono<ConsumerDmaapModel> findPnfinAAI(final ConsumerDmaapModel model) {
+
+ return getPnfModelClient
+ .getAaiResponse(model)
+ .flatMap(aaiModel -> Mono.just(model));
+ }
+
+
+
private Mono<AaiServiceInstanceQueryModel> checkIfPnfHasRelationToService(final AaiPnfResultModel model) {
+
return Mono
.justOrEmpty(model.getRelationshipList())
.map(this::findRelatedTo)
@@ -88,10 +107,12 @@ public class AaiQueryTaskImpl implements AaiQueryTask {
}
private Boolean checkIfRelatedServiceInstanceIsActive(final AaiServiceInstanceResultModel model) {
+
return ACTIVE_STATUS.equalsIgnoreCase(model.getOrchestrationStatus());
}
private Optional<RelationshipDict> findRelatedTo(final Relationship data) {
+
return Optional.ofNullable(data.getRelationship())
.map(Stream::of)
.orElseGet(Stream::empty)
@@ -101,10 +122,12 @@ public class AaiQueryTaskImpl implements AaiQueryTask {
}
private Optional<String> findValue(final List<RelationshipData> data, final String key) {
+
return data
.stream()
.filter(y -> key.equals(y.getRelationshipKey()))
.findFirst()
.map(RelationshipData::getRelationshipValue);
}
+
}
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 68a44ebc..b7c5c7ea 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
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -20,45 +21,51 @@
package org.onap.dcaegen2.services.prh.tasks;
-import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.INSTANCE_UUID;
-import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.RESPONSE_CODE;
-
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.CountDownLatch;
-import java.util.function.Predicate;
import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.exceptions.DmaapEmptyResponseException;
import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
import org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables;
+import org.slf4j.Marker;
+import org.slf4j.MarkerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
-import org.slf4j.Marker;
-import org.slf4j.MarkerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Profile;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.function.Predicate;
+import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.INSTANCE_UUID;
+import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.RESPONSE_CODE;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
*/
+/**
+ * @author <a href="mailto:sangeeta.bellara@t-systems.com">Sangeeta Bellara</a> on 3/12/23
+ */
+
+@Profile("!autoCommitDisabled")
@Component
public class ScheduledTasks {
private static final Logger LOGGER = LoggerFactory.getLogger(ScheduledTasks.class);
private static final Marker INVOKE = MarkerFactory.getMarker("INVOKE");
-
- private final DmaapConsumerTask dmaapConsumerTask;
- private final DmaapPublisherTask dmaapReadyProducerTask;
- private final DmaapPublisherTask dmaapUpdateProducerTask;
- private final AaiQueryTask aaiQueryTask;
- private final AaiProducerTask aaiProducerTask;
- private final BbsActionsTask bbsActionsTask;
+ private static Boolean pnfFound = true;
+ private DmaapConsumerTask dmaapConsumerTask;
+
+ private DmaapPublisherTask dmaapReadyProducerTask;
+ private DmaapPublisherTask dmaapUpdateProducerTask;
+ private AaiQueryTask aaiQueryTask;
+ private AaiProducerTask aaiProducerTask;
+ private BbsActionsTask bbsActionsTask;
private Map<String, String> mdcContextMap;
/**
@@ -69,6 +76,7 @@ public class ScheduledTasks {
* @param dmaapUpdatePublisherTask - fourth task
* @param aaiPublisherTask - second task
*/
+
@Autowired
public ScheduledTasks(
final DmaapConsumerTask dmaapConsumerTask,
@@ -90,8 +98,8 @@ public class ScheduledTasks {
static class State {
public final ConsumerDmaapModel dmaapModel;
public final Boolean activationStatus;
-
- public State(final ConsumerDmaapModel dmaapModel, final Boolean activationStatus) {
+
+ public State(ConsumerDmaapModel dmaapModel, final Boolean activationStatus) {
this.dmaapModel = dmaapModel;
this.activationStatus = activationStatus;
}
@@ -139,7 +147,7 @@ public class ScheduledTasks {
private void onError(Throwable throwable) {
if (!(throwable instanceof DmaapEmptyResponseException)) {
- LOGGER.warn("Chain of tasks have been aborted due to errors in PRH workflow", throwable);
+ LOGGER.warn("Chain of tasks have been aborted due to errors in PRH workflow {}", throwable);
}
}
@@ -153,7 +161,8 @@ public class ScheduledTasks {
}
private Mono<State> queryAaiForConfiguration(final ConsumerDmaapModel monoDMaaPModel) {
- return aaiQueryTask
+ LOGGER.info("Find AAI Info --> "+monoDMaaPModel.getCorrelationId());
+ return aaiQueryTask
.execute(monoDMaaPModel)
.map(x -> new State(monoDMaaPModel, x));
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java
index 70c54a51..5a5eb075 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java
@@ -3,6 +3,7 @@
* PNF-REGISTRATION-HANDLER
* ================================================================================
* Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2023 Deutsche Telekom 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.
@@ -23,44 +24,41 @@ package org.onap.dcaegen2.services.prh.tasks;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledFuture;
-
import javax.annotation.PreDestroy;
+import org.onap.dcaegen2.services.prh.configuration.CbsConfiguration;
import org.onap.dcaegen2.services.prh.configuration.PrhProperties;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.slf4j.Marker;
-import org.slf4j.MarkerFactory;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.annotation.EnableScheduling;
-
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 6/13/18
*/
+
+@Profile("!autoCommitDisabled")
@Configuration
@EnableScheduling
public class ScheduledTasksRunner {
- private static final Logger LOGGER = LoggerFactory.getLogger(ScheduledTasksRunner.class);
- private static final Marker ENTRY = MarkerFactory.getMarker("ENTRY");
- private static volatile List<ScheduledFuture> scheduledPrhTaskFutureList = new ArrayList<>();
+ private static volatile List<ScheduledFuture> scheduledPrhTaskFutureList = new ArrayList<>();
private final TaskScheduler taskScheduler;
private final ScheduledTasks scheduledTask;
private final PrhProperties prhProperties;
-
+ private final CbsConfiguration cbsConfiguration;
public ScheduledTasksRunner(TaskScheduler taskScheduler, ScheduledTasks scheduledTask,
- PrhProperties prhProperties) {
+ PrhProperties prhProperties, CbsConfiguration cbsConfiguration) {
this.taskScheduler = taskScheduler;
this.scheduledTask = scheduledTask;
this.prhProperties = prhProperties;
+ this.cbsConfiguration = cbsConfiguration;
}
-
+
@EventListener
public void onApplicationStartedEvent(ApplicationStartedEvent applicationStartedEvent) {
- tryToStartTask();
+ tryToStartTask();
}
/**
@@ -78,7 +76,6 @@ public class ScheduledTasksRunner {
* @return status of operation execution: true - started, false - not started
*/
public synchronized boolean tryToStartTask() {
- LOGGER.info(ENTRY, "Start scheduling PRH workflow");
if (scheduledPrhTaskFutureList.isEmpty()) {
scheduledPrhTaskFutureList.add(taskScheduler
.scheduleWithFixedDelay(scheduledTask::scheduleMainPrhEventTask,
@@ -88,5 +85,12 @@ public class ScheduledTasksRunner {
return false;
}
}
-
+
+ /**
+ * Function for cleaning resources for kafka subscriber and publisher.
+ */
+ public synchronized void closeKafkaPublisherSubscriber() {
+ cbsConfiguration.getMessageRouterSubscriber().close();
+ cbsConfiguration.getMessageRouterPublisher().close();
+ }
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/EpochDateTimeConversion.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/EpochDateTimeConversion.java
new file mode 100644
index 00000000..4bf49208
--- /dev/null
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/EpochDateTimeConversion.java
@@ -0,0 +1,95 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2023 Deutsche Telekom 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.tasks.commit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Date;
+
+/**
+ * This class will return start date time of the day and end date time of the day in epoch format.
+ * @author <a href="mailto:mohd.khan@t-systems.com">Mohd Usman Khan</a> on 3/13/23
+ */
+
+@Component
+public class EpochDateTimeConversion {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EpochDateTimeConversion.class);
+
+ private String daysForRecords = System.getenv("number_of_days");
+
+ public Long getStartDateOfTheDay(){
+ return getEpochDateTime(atStartOfDay(getCurrentDate()));
+ }
+
+ public Long getEndDateOfTheDay(){
+ return getEpochDateTime(atEndOfDay(getCurrentDate()));
+ }
+
+ private Long getEpochDateTime(Date date)
+ {
+ DateTimeFormatter dtf = DateTimeFormatter.ofPattern("E MMM dd HH:mm:ss zzz yyyy");
+ ZonedDateTime zdt = ZonedDateTime.parse( date.toString(),dtf);
+ return zdt.toInstant().toEpochMilli();
+ }
+
+ private Date getCurrentDate()
+ {
+ return new java.util.Date(System.currentTimeMillis());
+ }
+
+ public Date atStartOfDay(Date date) {
+ LocalDateTime localDateTime = dateToLocalDateTime(date);
+ if(daysForRecords==null)
+ daysForRecords="1";
+ LocalDateTime previousDay = localDateTime.minusDays(Integer.parseInt(daysForRecords) - 1l);
+ LocalDateTime previousStartTime = previousDay.with(LocalTime.MIN);
+ return localDateTimeToDate(previousStartTime);
+ }
+
+ private Date atEndOfDay(Date date) {
+ LocalDateTime localDateTime = dateToLocalDateTime(date);
+ LocalDateTime endOfDay = localDateTime.with(LocalTime.MAX);
+ return localDateTimeToDate(endOfDay);
+ }
+
+ private LocalDateTime dateToLocalDateTime(Date date) {
+ return LocalDateTime.ofInstant(date.toInstant(), ZoneId.systemDefault());
+ }
+
+ private Date localDateTimeToDate(LocalDateTime localDateTime) {
+ return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant());
+ }
+
+ public String getDaysForRecords() {
+ return daysForRecords;
+ }
+
+ public void setDaysForRecords(String daysForRecords) {
+ this.daysForRecords = daysForRecords;
+ }
+}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTask.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTask.java
new file mode 100644
index 00000000..4c70c713
--- /dev/null
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTask.java
@@ -0,0 +1,35 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2023 Deutsche Telekom 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.tasks.commit;
+
+import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel;
+import org.springframework.boot.configurationprocessor.json.JSONException;
+import reactor.core.publisher.Flux;
+
+/**
+ * @author <a href="mailto:ajinkya-patil@t-systems.com">Ajinkya Patil</a> on 3/13/23
+ */
+
+public interface KafkaConsumerTask {
+ Flux<ConsumerDmaapModel> execute() throws JSONException;
+
+ void commitOffset();
+}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImpl.java
new file mode 100644
index 00000000..6b289f1c
--- /dev/null
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImpl.java
@@ -0,0 +1,130 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2023 Deutsche Telekom 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.tasks.commit;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel;
+import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode;
+import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser;
+import org.springframework.boot.configurationprocessor.json.JSONException;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author <a href="mailto:ajinkya-patil@t-systems.com">Ajinkya Patil</a> on
+ * 3/13/23
+ */
+
+@Profile("autoCommitDisabled")
+@Component
+public class KafkaConsumerTaskImpl implements KafkaConsumerTask, BatchAcknowledgingMessageListener<String, String> {
+
+
+ private DmaapConsumerJsonParser dmaapConsumerJsonParser;
+
+ private EpochDateTimeConversion epochDateTimeConversion;
+
+ private CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode;
+
+ private List<String> jsonEvent = new ArrayList<>();
+
+ public List<String> getJsonEvent() {
+ return jsonEvent;
+ }
+
+ private Acknowledgment offset;
+
+ public Acknowledgment getOffset() {
+ return offset;
+ }
+
+ static String commonInURL = "/events/";
+
+ String kafkaTopic;
+
+ String groupIdConfig;
+
+
+ public KafkaConsumerTaskImpl(CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode
+ ,DmaapConsumerJsonParser dmaapConsumerJsonParser,EpochDateTimeConversion epochDateTimeConversion) {
+ this.cbsConfigurationForAutoCommitDisabledMode = cbsConfigurationForAutoCommitDisabledMode;
+ this.dmaapConsumerJsonParser = dmaapConsumerJsonParser;
+ this.epochDateTimeConversion = epochDateTimeConversion;
+ String kafkaTopicURL = this.cbsConfigurationForAutoCommitDisabledMode.getMessageRouterSubscribeRequest()
+ .sourceDefinition().topicUrl();
+ kafkaTopic = getTopicFromTopicUrl(kafkaTopicURL);
+ groupIdConfig = cbsConfigurationForAutoCommitDisabledMode.getMessageRouterSubscribeRequest().consumerGroup();
+
+ System.setProperty("kafkaTopic", kafkaTopic);
+ System.setProperty("groupIdConfig", groupIdConfig);
+
+ }
+
+ @Override
+ @KafkaListener(topics = "${kafkaTopic}", groupId = "${groupIdConfig}")
+ public void onMessage(List<ConsumerRecord<String, String>> list, Acknowledgment acknowledgment) {
+
+ if (list != null && !list.isEmpty()) {
+ list.stream().filter(
+ consumerRecord -> consumerRecord.timestamp() >= epochDateTimeConversion.getStartDateOfTheDay()
+ && consumerRecord.timestamp() <= epochDateTimeConversion.getEndDateOfTheDay())
+ .map(ConsumerRecord::value).forEach(value -> {
+ jsonEvent.add(value);
+ });
+
+ }
+
+ offset = acknowledgment;
+ }
+
+ @Override
+ public Flux<ConsumerDmaapModel> execute() throws JSONException {
+ return dmaapConsumerJsonParser.getConsumerDmaapModelFromKafkaConsumerRecord(jsonEvent);
+ }
+
+ public void setJsonEvent(List<String> jsonEvent) {
+ this.jsonEvent = jsonEvent;
+ }
+
+ @Override
+ public void commitOffset() {
+ if (!jsonEvent.isEmpty()) {
+ jsonEvent.clear();
+ }
+ if (offset != null) {
+ offset.acknowledge();
+ }
+ }
+
+ public String getTopicFromTopicUrl(String topicUrl) {
+ if (topicUrl.endsWith("/")) {
+ return topicUrl.substring(topicUrl.indexOf(commonInURL) + commonInURL.length(), topicUrl.lastIndexOf("/"));
+ }
+ return topicUrl.substring(topicUrl.indexOf(commonInURL) + commonInURL.length());
+ }
+
+}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksRunnerWithCommit.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksRunnerWithCommit.java
new file mode 100644
index 00000000..91cdd122
--- /dev/null
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksRunnerWithCommit.java
@@ -0,0 +1,98 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2023 Deutsche Telekom 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.tasks.commit;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ScheduledFuture;
+import javax.annotation.PreDestroy;
+import org.onap.dcaegen2.services.prh.configuration.PrhProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.Marker;
+import org.slf4j.MarkerFactory;
+import org.springframework.boot.context.event.ApplicationStartedEvent;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+/**
+ * @author <a href="mailto:pravin.kokane@t-systems.com">Pravin Kokane</a> on 3/13/23
+ */
+
+@Profile("autoCommitDisabled")
+@Configuration
+@EnableScheduling
+public class ScheduledTasksRunnerWithCommit {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ScheduledTasksRunnerWithCommit.class);
+ private static final Marker ENTRY = MarkerFactory.getMarker("ENTRY");
+ private static List<ScheduledFuture> scheduledPrhTaskFutureList = new ArrayList<>();
+
+ private final TaskScheduler taskScheduler;
+ private final PrhProperties prhProperties;
+
+ private ScheduledTasksWithCommit scheduledTasksWithCommit;
+
+ public ScheduledTasksRunnerWithCommit(TaskScheduler taskScheduler, ScheduledTasksWithCommit scheduledTasksWithCommit,
+ PrhProperties prhProperties) {
+ this.taskScheduler = taskScheduler;
+ this.scheduledTasksWithCommit = scheduledTasksWithCommit;
+ this.prhProperties = prhProperties;
+ }
+
+ @EventListener
+ public void onApplicationStartedEvent(ApplicationStartedEvent applicationStartedEvent) {
+ LOGGER.info(ENTRY,"### in onApplicationStartedEvent");
+ LOGGER.info(ENTRY,"###tryToStartTaskWithCommit="+tryToStartTaskWithCommit());
+ }
+
+ /**
+ * Function which have to stop tasks execution.
+ */
+ @PreDestroy
+ public synchronized void cancelTasks() {
+ LOGGER.info(ENTRY,"###In cancelTasks");
+ scheduledPrhTaskFutureList.forEach(x -> x.cancel(false));
+ scheduledPrhTaskFutureList.clear();
+ }
+
+ /**
+ * Function for starting scheduling PRH workflow.
+ *
+ * @return status of operation execution: true - started, false - not started
+ */
+
+ public synchronized boolean tryToStartTaskWithCommit() {
+ LOGGER.info(ENTRY, "Start scheduling PRH workflow with Commit Tasks Runner");
+ if (scheduledPrhTaskFutureList.isEmpty()) {
+ Collections.synchronizedList(scheduledPrhTaskFutureList);
+ scheduledPrhTaskFutureList.add(taskScheduler
+ .scheduleWithFixedDelay(scheduledTasksWithCommit::scheduleKafkaPrhEventTask,
+ prhProperties.getWorkflowSchedulingInterval()));
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksWithCommit.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksWithCommit.java
new file mode 100644
index 00000000..352c0bbc
--- /dev/null
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksWithCommit.java
@@ -0,0 +1,203 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2023 Deutsche Telekom 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.tasks.commit;
+
+import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.RESPONSE_CODE;
+
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import org.onap.dcaegen2.services.prh.exceptions.DmaapEmptyResponseException;
+import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
+import org.onap.dcaegen2.services.prh.tasks.AaiProducerTask;
+import org.onap.dcaegen2.services.prh.tasks.AaiQueryTask;
+import org.onap.dcaegen2.services.prh.tasks.BbsActionsTask;
+import org.onap.dcaegen2.services.prh.tasks.DmaapPublisherTask;
+import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.configurationprocessor.json.JSONException;
+import org.springframework.context.annotation.Profile;
+import org.springframework.http.HttpStatus;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author <a href="mailto:sangeeta.bellara@t-systems.com">Sangeeta Bellara</a>
+ * on 3/13/23
+ */
+@Profile("autoCommitDisabled")
+@Component
+public class ScheduledTasksWithCommit {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ScheduledTasksWithCommit.class);
+ private static Boolean pnfFound = true;
+ private KafkaConsumerTask kafkaConsumerTask;
+ private DmaapPublisherTask dmaapReadyProducerTask;
+ private DmaapPublisherTask dmaapUpdateProducerTask;
+ public AaiQueryTask aaiQueryTask;
+ private AaiProducerTask aaiProducerTask;
+ private BbsActionsTask bbsActionsTask;
+ private Map<String, String> mdcContextMap;
+
+ /**
+ * Constructor for tasks registration in PRHWorkflow.
+ *
+ * @param kafkaConsumerTask - fist task
+ * @param dmaapReadyPublisherTask - third task
+ * @param dmaapUpdatePublisherTask - fourth task
+ * @param aaiPublisherTask - second task
+ */
+ @Autowired
+ public ScheduledTasksWithCommit(final KafkaConsumerTask kafkaConsumerTask,
+ @Qualifier("ReadyPublisherTask") final DmaapPublisherTask dmaapReadyPublisherTask,
+ @Qualifier("UpdatePublisherTask") final DmaapPublisherTask dmaapUpdatePublisherTask,
+ final AaiQueryTask aaiQueryTask, final AaiProducerTask aaiPublisherTask,
+ final BbsActionsTask bbsActionsTask, final Map<String, String> mdcContextMap)
+
+ {
+ this.dmaapReadyProducerTask = dmaapReadyPublisherTask;
+ this.dmaapUpdateProducerTask = dmaapUpdatePublisherTask;
+ this.kafkaConsumerTask = kafkaConsumerTask;
+ this.aaiQueryTask = aaiQueryTask;
+ this.aaiProducerTask = aaiPublisherTask;
+ this.bbsActionsTask = bbsActionsTask;
+ this.mdcContextMap = mdcContextMap;
+ }
+
+ static class State {
+ public ConsumerDmaapModel dmaapModel;
+ public Boolean activationStatus;
+
+ public State(ConsumerDmaapModel dmaapModel, final Boolean activationStatus) {
+ this.dmaapModel = dmaapModel;
+ this.activationStatus = activationStatus;
+ }
+ }
+
+ public void scheduleKafkaPrhEventTask() {
+ MdcVariables.setMdcContextMap(mdcContextMap);
+ try {
+
+ LOGGER.info("Execution of tasks was registered with commit");
+ CountDownLatch mainCountDownLatch = new CountDownLatch(1);
+ consumeFromKafkaMessage()
+ .flatMap(model -> queryAaiForPnf(model).doOnError(e -> {
+ LOGGER.info("PNF Not Found in AAI --> {}" + e);
+ LOGGER.info("PNF Not Found in AAI With description of exception --> {}" + e.getMessage());
+ disableCommit();
+ }).onErrorResume(e -> Mono.empty())
+
+ )
+ .flatMap(this::queryAaiForConfiguration)
+ .flatMap(this::publishToAaiConfiguration)
+ .flatMap(this::processAdditionalFields).flatMap(this::publishToDmaapConfiguration)
+
+ .onErrorResume(e -> Mono.empty())
+
+ .doOnTerminate(mainCountDownLatch::countDown)
+ .subscribe(this::onSuccess, this::onError, this::onCompleteKafka);
+ mainCountDownLatch.await();
+ } catch (InterruptedException | JSONException e) {
+ LOGGER.warn("Interruption problem on countDownLatch {}", e);
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private static void disableCommit() {
+ pnfFound = false;
+ }
+
+ private void onCompleteKafka() {
+ LOGGER.info("PRH tasks have been completed");
+ if (pnfFound) {
+ kafkaConsumerTask.commitOffset();
+ LOGGER.info("Committed the Offset");
+ } else {
+ LOGGER.info("Offset not Committed");
+ pnfFound = true;
+ }
+ }
+
+ private void onSuccess(MessageRouterPublishResponse response) {
+ if (response.successful()) {
+ String statusCodeOk = HttpStatus.OK.name();
+ MDC.put(RESPONSE_CODE, statusCodeOk);
+ LOGGER.info("Prh consumed tasks successfully. HTTP Response code from DMaaPProducer {}", statusCodeOk);
+ MDC.remove(RESPONSE_CODE);
+ }
+ }
+
+ private void onError(Throwable throwable) {
+ if (!(throwable instanceof DmaapEmptyResponseException)) {
+ LOGGER.warn("Chain of tasks have been aborted due to errors in PRH workflow {}", throwable);
+ }
+ }
+
+ private Flux<ConsumerDmaapModel> consumeFromKafkaMessage() throws JSONException {
+ return kafkaConsumerTask.execute();
+ }
+
+ private Mono<State> queryAaiForConfiguration(final ConsumerDmaapModel monoDMaaPModel) {
+ return aaiQueryTask.execute(monoDMaaPModel).map(x -> new State(monoDMaaPModel, x));
+ }
+
+ private Mono<ConsumerDmaapModel> queryAaiForPnf(final ConsumerDmaapModel monoDMaaPModel) {
+
+ LOGGER.info("Find PNF --> " + monoDMaaPModel.getCorrelationId());
+ return aaiQueryTask.findPnfinAAI(monoDMaaPModel);
+ }
+
+ private Mono<State> publishToAaiConfiguration(final State state) {
+ try {
+ return aaiProducerTask.execute(state.dmaapModel).map(x -> state);
+ } catch (PrhTaskException e) {
+ LOGGER.warn("AAIProducerTask exception has been registered: {}", e);
+ return Mono.error(e);
+ }
+ }
+
+ private Mono<State> processAdditionalFields(final State state) {
+ if (state.activationStatus) {
+ LOGGER.debug("Re-registration - Logical links won't be updated.");
+ return Mono.just(state);
+ }
+ return bbsActionsTask.execute(state.dmaapModel).map(x -> state);
+ }
+
+ private Flux<MessageRouterPublishResponse> publishToDmaapConfiguration(final State state) {
+ try {
+ if (state.activationStatus) {
+ LOGGER.debug("Re-registration - Using PNF_UPDATE DMaaP topic.");
+ return dmaapUpdateProducerTask.execute(state.dmaapModel);
+ }
+ return dmaapReadyProducerTask.execute(state.dmaapModel);
+ } catch (PrhTaskException e) {
+ LOGGER.warn("DMaaPProducerTask exception has been registered: ", e);
+ return Flux.error(e);
+ }
+ }
+}
diff --git a/prh-app-server/src/main/resources/application.yaml b/prh-app-server/src/main/resources/application.yaml
index 8f1950d0..e62d4e90 100644
--- a/prh-app-server/src/main/resources/application.yaml
+++ b/prh-app-server/src/main/resources/application.yaml
@@ -1,6 +1,6 @@
spring:
- profiles:
- active: prod
+ profiles: prod
+
server:
port: 8433
ssl:
@@ -25,4 +25,13 @@ logging:
management.endpoints.web.exposure.include: "*"
+---
+spring:
+ profiles:
+ default: prod
+
+
+
+
+
diff --git a/prh-app-server/src/main/resources/logback-spring.xml b/prh-app-server/src/main/resources/logback-spring.xml
index 060cf6c5..c4d31e52 100644
--- a/prh-app-server/src/main/resources/logback-spring.xml
+++ b/prh-app-server/src/main/resources/logback-spring.xml
@@ -1,8 +1,27 @@
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ ============LICENSE_START=======================================================
+ ~ PNF-REGISTRATION-HANDLER
+ ~ ================================================================================
+ ~ Copyright (C) 2023 Deutsche Telekom 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=========================================================
+-->
<configuration debug="false" scan="false">
<include resource="org/springframework/boot/logging/logback/defaults.xml"/>
<property name="outputFilename" value="prh-app-server_output"/>
- <property name="logPath" value="/var/log/ONAP/prh/prh-app-server"/>
+ <property name="logPath" value="/var/log/ONAP/prh/prh-app-server"/>
<property name="archivePath" value="${logPath}/archive"/>
<property name="maxFileSize" value="50MB"/>
<property name="maxHistory" value="30"/>
@@ -17,7 +36,7 @@
|%thread
|%n"/>
- <springProfile name="prod">
+ <springProfile name="prod,autoCommitDisabled">
<appender class="ch.qos.logback.core.ConsoleAppender" name="CONSOLE" target="SYSTEM_OUT">
<encoder>
<pattern>${defaultPattern}</pattern>