diff options
Diffstat (limited to 'prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration')
5 files changed, 172 insertions, 84 deletions
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..64fff9a7 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,35 @@ 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()); messageRouterCBSSubscribeRequest = consulConfigurationParser.getMessageRouterSubscribeRequest(); - } - + } @Override public MessageRouterPublisher getMessageRouterPublisher() { @@ -72,21 +73,27 @@ 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 index 8affe281..baaf3b16 100644 --- 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 @@ -17,80 +17,90 @@ * 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.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.context.annotation.Profile; 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; -import java.util.HashMap; -import java.util.Map; - -/** - * @author <a href="mailto:pravin.kokane@t-systems.com">Pravin Kokane</a> on 3/13/23 - */ + /** + * * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on + * * 24/08/23 + * */ @Profile("autoCommitDisabled") @EnableKafka @Configuration -public class KafkaConfig -{ - String kafkaBoostrapServerConfig = System.getenv("kafkaBoostrapServerConfig"); - - String groupIdConfig = System.getenv("groupIdConfig"); - - - String kafkaSecurityProtocol = System.getenv("kafkaSecurityProtocol"); - - String kafkaSaslMechanism = System.getenv("kafkaSaslMechanism"); - - String kafkaUsername = System.getenv("kafkaUsername"); - - String kafkaPassword = System.getenv("kafkaPassword"); - - String kafkaJaasConfig = System.getenv("JAAS_CONFIG"); - - String kafkaLoginModuleClassConfig = System.getenv("Login_Module_Class"); +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() - { - 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"); + 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(kafkaJaasConfig == null) { - kafkaJaasConfig = kafkaLoginModuleClassConfig + " required username=\"" - + kafkaUsername + "\" password=\"" + kafkaPassword + "\";"; - } - if(kafkaSecurityProtocol==null ) kafkaSecurityProtocol="SASL_PLAINTEXT"; + + if (kafkaSecurityProtocol == null) + kafkaSecurityProtocol = DEFAULT_KAFKA_SECURITY_PROTOCOL; config.put("security.protocol", kafkaSecurityProtocol); - if(kafkaSaslMechanism==null ) kafkaSaslMechanism="SCRAM-SHA-512"; + 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 kafkaListenerContainerFactory() - { + public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory( + CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode) { ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); - factory.setConsumerFactory(consumerFactory()); + factory.setConsumerFactory(consumerFactory(cbsConfigurationForAutoCommitDisabledMode)); factory.setBatchListener(true); factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL); return factory; } + } |