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.Optional;
  22 import java.util.Properties;
  23 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.EnvProperties;
  24 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.ReactiveCloudConfigurationProvider;
  25 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
  26 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
  27 import org.slf4j.Logger;
  28 import org.slf4j.LoggerFactory;
  29 import org.springframework.beans.factory.annotation.Autowired;
  30 import org.springframework.beans.factory.annotation.Value;
  31 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  32 import org.springframework.context.annotation.ComponentScan;
  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  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  42  */
  43 @Configuration
  44 @ComponentScan("org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers")
  45 @EnableConfigurationProperties
  46 @EnableScheduling
  47 @Primary
  48 public class CloudConfiguration extends AppConfig {
  49     private static final Logger logger = LoggerFactory.getLogger(CloudConfiguration.class);
  50     private ReactiveCloudConfigurationProvider reactiveCloudConfigurationProvider;
  51     private DmaapPublisherConfiguration dmaapPublisherCloudConfiguration;
  52     private DmaapConsumerConfiguration dmaapConsumerCloudConfiguration;
  53     private FtpesConfig ftpesCloudConfiguration;
  54
  55     @Value("#{systemEnvironment}")
  56     private Properties systemEnvironment;
  57
  58     @Autowired
  59     public void setThreadPoolTaskScheduler(ReactiveCloudConfigurationProvider reactiveCloudConfigurationProvider) {
  60         this.reactiveCloudConfigurationProvider = reactiveCloudConfigurationProvider;
  61     }
  62
  63
  64     protected void runTask(Map<String, String> contextMap) {
  65         Flux.defer(() -> EnvironmentProcessor.evaluate(systemEnvironment, contextMap)).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         reactiveCloudConfigurationProvider.callForServiceConfigurationReactive(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         ftpesCloudConfiguration = cloudConfigParser.getFtpesConfig();
  89     }
  90
  91     @Override
  92     public DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
  93         return Optional.ofNullable(dmaapPublisherCloudConfiguration).orElse(super.getDmaapPublisherConfiguration());
  94     }
  95
  96     @Override
  97     public DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
  98         return Optional.ofNullable(dmaapConsumerCloudConfiguration).orElse(super.getDmaapConsumerConfiguration());
  99     }
 100
 101     @Override
 102     public FtpesConfig getFtpesConfiguration() {
 103         return Optional.ofNullable(ftpesCloudConfiguration).orElse(super.getFtpesConfiguration());
 104     }
 105 }