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

   1 /*
   2  * ============LICENSE_START======================================================================
   3  * Copyright (C) 2018 NOKIA Intellectual Property, 2018 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.Optional;
  22 import java.util.Properties;
  23
  24 import org.onap.dcaegen2.collectors.datafile.config.DmaapConsumerConfiguration;
  25 import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
  26 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.EnvProperties;
  27 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.ReactiveCloudConfigurationProvider;
  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.Flux;
  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 void setThreadPoolTaskScheduler(ReactiveCloudConfigurationProvider reactiveCloudConfigurationProvider) {
  62         this.reactiveCloudConfigurationProvider = reactiveCloudConfigurationProvider;
  63     }
  64
  65
  66     protected void runTask() {
  67         Flux.defer(() -> EnvironmentProcessor.evaluate(systemEnvironment)).subscribeOn(Schedulers.parallel())
  68                 .subscribe(this::parsingConfigSuccess, this::parsingConfigError);
  69     }
  70
  71     private void parsingConfigError(Throwable throwable) {
  72         logger.warn("Error in case of processing system environment, more details below: ", throwable);
  73     }
  74
  75     private void cloudConfigError(Throwable throwable) {
  76         logger.warn("Exception during getting configuration from CONSUL/CONFIG_BINDING_SERVICE ", throwable);
  77     }
  78
  79     private void parsingConfigSuccess(EnvProperties envProperties) {
  80         logger.info("Fetching Datafile Collector configuration from ConfigBindingService/Consul");
  81         reactiveCloudConfigurationProvider.callForServiceConfigurationReactive(envProperties)
  82                 .subscribe(this::parseCloudConfig, this::cloudConfigError);
  83     }
  84
  85     private void 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     }
  92
  93     @Override
  94     public DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
  95         return Optional.ofNullable(dmaapPublisherCloudConfiguration).orElse(super.getDmaapPublisherConfiguration());
  96     }
  97
  98     @Override
  99     public DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
 100         return Optional.ofNullable(dmaapConsumerCloudConfiguration).orElse(super.getDmaapConsumerConfiguration());
 101     }
 102
 103     @Override
 104     public FtpesConfig getFtpesConfiguration() {
 105         return Optional.ofNullable(ftpesCloudConfiguration).orElse(super.getFtpesConfiguration());
 106     }
 107 }