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 import java.util.Map;
  21 import java.util.Properties;
  22 import org.onap.dcaegen2.collectors.datafile.model.logging.MappedDiagnosticContext;
  23 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.ReactiveCloudConfigurationProvider;
  24 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
  25 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28 import org.springframework.beans.factory.annotation.Autowired;
  29 import org.springframework.beans.factory.annotation.Value;
  30 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  31 import org.springframework.context.annotation.ComponentScan;
  32 import org.springframework.context.annotation.Configuration;
  33 import org.springframework.context.annotation.Primary;
  34 import org.springframework.scheduling.annotation.EnableScheduling;
  35 import reactor.core.publisher.Mono;
  36 import reactor.core.scheduler.Schedulers;
  37
  38 /**
  39  * Gets the DFC configuration from the ConfigBindingService/Consul and parses it to the configurations needed in DFC.
  40  *
  41  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 9/19/18
  42  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  43  */
  44 @Configuration
  45 @ComponentScan("org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers")
  46 @EnableConfigurationProperties
  47 @EnableScheduling
  48 @Primary
  49 public class CloudConfiguration extends AppConfig {
  50     private static final Logger logger = LoggerFactory.getLogger(CloudConfiguration.class);
  51     private ReactiveCloudConfigurationProvider reactiveCloudConfigurationProvider;
  52     private DmaapPublisherConfiguration dmaapPublisherCloudConfiguration;
  53     private DmaapConsumerConfiguration dmaapConsumerCloudConfiguration;
  54     private FtpesConfig ftpesCloudConfiguration;
  55
  56     @Value("#{systemEnvironment}")
  57     private Properties systemEnvironment;
  58
  59     @Autowired
  60     public synchronized void setThreadPoolTaskScheduler(
  61             ReactiveCloudConfigurationProvider reactiveCloudConfigurationProvider) {
  62         this.reactiveCloudConfigurationProvider = reactiveCloudConfigurationProvider;
  63     }
  64
  65     /**
  66      * Reads the cloud configuration.
  67      */
  68     public void runTask() {
  69         Map<String,String> context = MappedDiagnosticContext.initializeTraceContext();
  70         EnvironmentProcessor.readEnvironmentVariables(systemEnvironment, context) //
  71                 .subscribeOn(Schedulers.parallel()) //
  72                 .flatMap(reactiveCloudConfigurationProvider::callForServiceConfigurationReactive) //
  73                 .flatMap(this::parseCloudConfig) //
  74                 .subscribe(null, this::onError, this::onComplete);
  75     }
  76
  77     private void onComplete() {
  78         logger.trace("Configuration updated");
  79     }
  80
  81     private void onError(Throwable throwable) {
  82         logger.warn("Exception during getting configuration from CONSUL/CONFIG_BINDING_SERVICE ", throwable);
  83     }
  84
  85     private synchronized Mono<CloudConfiguration> parseCloudConfig(JsonObject jsonObject) {
  86         logger.info("Received application configuration: {}", jsonObject);
  87         CloudConfigParser cloudConfigParser = new CloudConfigParser(jsonObject);
  88         dmaapPublisherCloudConfiguration = cloudConfigParser.getDmaapPublisherConfig();
  89         dmaapConsumerCloudConfiguration = cloudConfigParser.getDmaapConsumerConfig();
  90         ftpesCloudConfiguration = cloudConfigParser.getFtpesConfig();
  91         return Mono.just(this);
  92     }
  93
  94     @Override
  95     public synchronized DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
  96         return dmaapPublisherCloudConfiguration != null ? dmaapPublisherCloudConfiguration
  97                 : super.getDmaapPublisherConfiguration();
  98     }
  99
 100     @Override
 101     public synchronized DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
 102         return dmaapConsumerCloudConfiguration != null ? dmaapConsumerCloudConfiguration
 103                 : super.getDmaapConsumerConfiguration();
 104     }
 105
 106     @Override
 107     public synchronized FtpesConfig getFtpesConfiguration() {
 108         return ftpesCloudConfiguration != null ? ftpesCloudConfiguration : super.getFtpesConfiguration();
 109     }
 110 }