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.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
  25 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
  26
  27 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.EnvProperties;
  28 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.ReactiveCloudConfigurationProvider;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31 import org.springframework.beans.factory.annotation.Autowired;
  32 import org.springframework.beans.factory.annotation.Value;
  33 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  34 import org.springframework.context.annotation.ComponentScan;
  35 import org.springframework.context.annotation.Configuration;
  36 import org.springframework.context.annotation.Primary;
  37 import org.springframework.scheduling.annotation.EnableScheduling;
  38
  39 import reactor.core.publisher.Flux;
  40 import reactor.core.scheduler.Schedulers;
  41
  42 /**
  43  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 9/19/18
  44  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  45  */
  46 @Configuration
  47 @ComponentScan("org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers")
  48 @EnableConfigurationProperties
  49 @EnableScheduling
  50 @Primary
  51 public class CloudConfiguration extends AppConfig {
  52     private static final Logger logger = LoggerFactory.getLogger(CloudConfiguration.class);
  53     private ReactiveCloudConfigurationProvider reactiveCloudConfigurationProvider;
  54     private DmaapPublisherConfiguration dmaapPublisherCloudConfiguration;
  55     private DmaapConsumerConfiguration dmaapConsumerCloudConfiguration;
  56     private FtpesConfig ftpesCloudConfiguration;
  57
  58     @Value("#{systemEnvironment}")
  59     private Properties systemEnvironment;
  60
  61     @Autowired
  62     public void setThreadPoolTaskScheduler(ReactiveCloudConfigurationProvider reactiveCloudConfigurationProvider) {
  63         this.reactiveCloudConfigurationProvider = reactiveCloudConfigurationProvider;
  64     }
  65
  66
  67     protected void runTask() {
  68         Flux.defer(() -> EnvironmentProcessor.evaluate(systemEnvironment)).subscribeOn(Schedulers.parallel())
  69                 .subscribe(this::parsingConfigSuccess, this::parsingConfigError);
  70     }
  71
  72     private void parsingConfigError(Throwable throwable) {
  73         logger.warn("Error in case of processing system environment, more details below: ", throwable);
  74     }
  75
  76     private void cloudConfigError(Throwable throwable) {
  77         logger.warn("Exception during getting configuration from CONSUL/CONFIG_BINDING_SERVICE ", throwable);
  78     }
  79
  80     private void parsingConfigSuccess(EnvProperties envProperties) {
  81         logger.info("Fetching Datafile Collector configuration from ConfigBindingService/Consul");
  82         reactiveCloudConfigurationProvider.callForServiceConfigurationReactive(envProperties)
  83                 .subscribe(this::parseCloudConfig, this::cloudConfigError);
  84     }
  85
  86     private void parseCloudConfig(JsonObject jsonObject) {
  87         logger.info("Received application configuration: {}", jsonObject);
  88         CloudConfigParser cloudConfigParser = new CloudConfigParser(jsonObject);
  89         dmaapPublisherCloudConfiguration = cloudConfigParser.getDmaapPublisherConfig();
  90         dmaapConsumerCloudConfiguration = cloudConfigParser.getDmaapConsumerConfig();
  91         ftpesCloudConfiguration = cloudConfigParser.getFtpesConfig();
  92     }
  93
  94     @Override
  95     public DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
  96         return Optional.ofNullable(dmaapPublisherCloudConfiguration).orElse(super.getDmaapPublisherConfiguration());
  97     }
  98
  99     @Override
 100     public DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
 101         return Optional.ofNullable(dmaapConsumerCloudConfiguration).orElse(super.getDmaapConsumerConfiguration());
 102     }
 103
 104     @Override
 105     public FtpesConfig getFtpesConfiguration() {
 106         return Optional.ofNullable(ftpesCloudConfiguration).orElse(super.getFtpesConfiguration());
 107     }
 108 }