gerrit.onap Code Review - dcaegen2/collectors/datafile.git/blob

   1 /*
   2  * ============LICENSE_START======================================================================
   3  * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 Nordix Foundation. All rights reserved.
   4  * ===============================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   6  * in compliance with the License. You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software distributed under the License
  11  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  12  * or implied. See the License for the specific language governing permissions and limitations under
  13  * the License.
  14  * ============LICENSE_END========================================================================
  15  */
  16
  17 package org.onap.dcaegen2.collectors.datafile.configuration;
  18
  19 import com.google.gson.JsonObject;
  20
  21 import java.util.Map;
  22 import java.util.Properties;
  23
  24 import org.onap.dcaegen2.collectors.datafile.model.logging.MappedDiagnosticContext;
  25 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.ReactiveCloudConfigurationProvider;
  26 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
  27 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30 import org.springframework.beans.factory.annotation.Autowired;
  31 import org.springframework.beans.factory.annotation.Value;
  32 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  33 import org.springframework.context.annotation.ComponentScan;
  34 import org.springframework.context.annotation.Configuration;
  35 import org.springframework.context.annotation.Primary;
  36 import org.springframework.scheduling.annotation.EnableScheduling;
  37
  38 import reactor.core.publisher.Mono;
  39 import reactor.core.scheduler.Schedulers;
  40
  41 /**
  42  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 9/19/18
  43  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  44  */
  45 @Configuration
  46 @ComponentScan("org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers")
  47 @EnableConfigurationProperties
  48 @EnableScheduling
  49 @Primary
  50 public class CloudConfiguration extends AppConfig {
  51     private static final Logger logger = LoggerFactory.getLogger(CloudConfiguration.class);
  52     private ReactiveCloudConfigurationProvider reactiveCloudConfigurationProvider;
  53     private DmaapPublisherConfiguration dmaapPublisherCloudConfiguration;
  54     private DmaapConsumerConfiguration dmaapConsumerCloudConfiguration;
  55     private FtpesConfig ftpesCloudConfiguration;
  56
  57     @Value("#{systemEnvironment}")
  58     private Properties systemEnvironment;
  59
  60     @Autowired
  61     public synchronized void setThreadPoolTaskScheduler(
  62             ReactiveCloudConfigurationProvider reactiveCloudConfigurationProvider) {
  63         this.reactiveCloudConfigurationProvider = reactiveCloudConfigurationProvider;
  64     }
  65
  66     public void runTask() {
  67         Map<String,String> context = MappedDiagnosticContext.initializeTraceContext();
  68         EnvironmentProcessor.readEnvironmentVariables(systemEnvironment, context) //
  69                 .subscribeOn(Schedulers.parallel()) //
  70                 .flatMap(reactiveCloudConfigurationProvider::callForServiceConfigurationReactive) //
  71                 .flatMap(this::parseCloudConfig) //
  72                 .subscribe(null, this::onError, this::onComplete);
  73     }
  74
  75     private void onComplete() {
  76         logger.trace("Configuration updated");
  77     }
  78
  79     private void onError(Throwable throwable) {
  80         logger.warn("Exception during getting configuration from CONSUL/CONFIG_BINDING_SERVICE ", throwable);
  81     }
  82
  83     private synchronized Mono<CloudConfiguration> parseCloudConfig(JsonObject jsonObject) {
  84         logger.info("Received application configuration: {}", jsonObject);
  85         CloudConfigParser cloudConfigParser = new CloudConfigParser(jsonObject);
  86         dmaapPublisherCloudConfiguration = cloudConfigParser.getDmaapPublisherConfig();
  87         dmaapConsumerCloudConfiguration = cloudConfigParser.getDmaapConsumerConfig();
  88         ftpesCloudConfiguration = cloudConfigParser.getFtpesConfig();
  89         return Mono.just(this);
  90     }
  91
  92     @Override
  93     public synchronized DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
  94         return dmaapPublisherCloudConfiguration != null ? dmaapPublisherCloudConfiguration
  95                 : super.getDmaapPublisherConfiguration();
  96     }
  97
  98     @Override
  99     public synchronized DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
 100         return dmaapConsumerCloudConfiguration != null ? dmaapConsumerCloudConfiguration
 101                 : super.getDmaapConsumerConfiguration();
 102     }
 103
 104     @Override
 105     public synchronized FtpesConfig getFtpesConfiguration() {
 106         return ftpesCloudConfiguration != null ? ftpesCloudConfiguration : super.getFtpesConfiguration();
 107     }
 108 }