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

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  * ============LICENSE_END=========================================================
  17  */
  18
  19 package org.onap.dcaegen2.collectors.datafile.configuration;
  20
  21 import com.google.gson.JsonObject;
  22 import java.util.Optional;
  23 import java.util.Properties;
  24 import org.onap.dcaegen2.collectors.datafile.config.DmaapConsumerConfiguration;
  25 import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
  26 import org.onap.dcaegen2.collectors.datafile.model.EnvProperties;
  27 import org.onap.dcaegen2.collectors.datafile.service.DatafileConfigurationProvider;
  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.Configuration;
  34 import org.springframework.context.annotation.Primary;
  35 import org.springframework.scheduling.annotation.EnableScheduling;
  36 import reactor.core.publisher.Flux;
  37 import reactor.core.scheduler.Schedulers;
  38
  39 /**
  40  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 9/19/18
  41  */
  42 @Configuration
  43 @EnableConfigurationProperties
  44 @EnableScheduling
  45 @Primary
  46 public class CloudConfiguration extends AppConfig {
  47
  48     private static final Logger logger = LoggerFactory.getLogger(CloudConfiguration.class);
  49
  50     private DatafileConfigurationProvider datafileConfigurationProvider;
  51     private DmaapPublisherConfiguration dmaapPublisherCloudConfiguration;
  52     private DmaapConsumerConfiguration dmaapConsumerCloudConfiguration;
  53
  54     @Value("#{systemEnvironment}")
  55     private Properties systemEnvironment;
  56
  57     @Autowired
  58     public void setThreadPoolTaskScheduler(DatafileConfigurationProvider datafileConfigurationProvider) {
  59         this.datafileConfigurationProvider = datafileConfigurationProvider;
  60     }
  61
  62
  63     protected void runTask() {
  64         Flux.defer(() -> EnvironmentProcessor.evaluate(systemEnvironment))
  65             .subscribeOn(Schedulers.parallel())
  66             .subscribe(this::parsingConfigSuccess, this::parsingConfigError);
  67     }
  68
  69     private void parsingConfigError(Throwable throwable) {
  70         logger.warn("Error in case of processing system environment, more details below: ", throwable);
  71     }
  72
  73     private void cloudConfigError(Throwable throwable) {
  74         logger.warn("Exception during getting configuration from CONSUL/CONFIG_BINDING_SERVICE ", throwable);
  75     }
  76
  77     private void parsingConfigSuccess(EnvProperties envProperties) {
  78         logger.info("Fetching Datafile Collector configuration from ConfigBindingService/Consul");
  79         datafileConfigurationProvider.callForDataFileCollectorConfiguration(envProperties)
  80             .subscribe(this::parseCloudConfig, this::cloudConfigError);
  81     }
  82
  83     private void 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     }
  89
  90     @Override
  91     public DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
  92         return Optional.ofNullable(dmaapPublisherCloudConfiguration).orElse(super.getDmaapPublisherConfiguration());
  93     }
  94
  95     @Override
  96     public DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
  97         return Optional.ofNullable(dmaapConsumerCloudConfiguration).orElse(super.getDmaapConsumerConfiguration());
  98     }
  99 }