enrichment-coordinator-service/src/main/java/org/oransc/enrichment/controllers/producer/ProducerCallbacks.java

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * O-RAN-SC
   4  * %%
   5  * Copyright (C) 2020 Nordix Foundation
   6  * %%
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ========================LICENSE_END===================================
  19  */
  20
  21 package org.oransc.enrichment.controllers.producer;
  22
  23 import com.google.gson.Gson;
  24 import com.google.gson.GsonBuilder;
  25
  26 import java.lang.invoke.MethodHandles;
  27 import java.time.Duration;
  28 import java.util.Collection;
  29
  30 import org.oransc.enrichment.clients.AsyncRestClient;
  31 import org.oransc.enrichment.clients.AsyncRestClientFactory;
  32 import org.oransc.enrichment.configuration.ApplicationConfig;
  33 import org.oransc.enrichment.repository.EiJob;
  34 import org.oransc.enrichment.repository.EiJobs;
  35 import org.oransc.enrichment.repository.EiProducer;
  36 import org.oransc.enrichment.repository.EiProducers;
  37 import org.slf4j.Logger;
  38 import org.slf4j.LoggerFactory;
  39
  40 import reactor.core.publisher.Flux;
  41 import reactor.core.publisher.Mono;
  42 import reactor.util.retry.Retry;
  43
  44 /**
  45  * Callbacks to the EiProducer
  46  */
  47 @SuppressWarnings("java:S3457") // No need to call "toString()" method as formatting and string ..
  48 public class ProducerCallbacks {
  49
  50     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  51     private static Gson gson = new GsonBuilder().create();
  52
  53     private final AsyncRestClient restClient;
  54
  55     public ProducerCallbacks(ApplicationConfig config) {
  56         AsyncRestClientFactory restClientFactory = new AsyncRestClientFactory(config.getWebClientConfig());
  57         this.restClient = restClientFactory.createRestClientNoHttpProxy("");
  58     }
  59
  60     public Mono<String> healthCheck(EiProducer producer) {
  61         return restClient.get(producer.getProducerSupervisionCallbackUrl());
  62     }
  63
  64     public void stopEiJob(EiJob eiJob, EiProducers eiProducers) {
  65         for (EiProducer producer : getProducersForJob(eiJob, eiProducers)) {
  66             String url = producer.getJobCallbackUrl() + "/" + eiJob.getId();
  67             producer.setJobDisabled(eiJob);
  68             restClient.delete(url) //
  69                 .subscribe(response -> logger.debug("Producer job deleted OK {}", producer.getId()), //
  70                     throwable -> logger.warn("Producer job delete failed {} {}", producer.getId(),
  71                         throwable.getMessage()),
  72                     null);
  73         }
  74     }
  75
  76     /**
  77      * Start a job in all producers that suports the job type
  78      *
  79      * @param eiJob an EI job
  80      * @return the number of producers that returned OK
  81      */
  82     public Mono<Integer> startEiJob(EiJob eiJob, EiProducers eiProducers) {
  83         Retry retrySpec = Retry.fixedDelay(1, Duration.ofSeconds(1));
  84         return Flux.fromIterable(getProducersForJob(eiJob, eiProducers)) //
  85             .flatMap(eiProducer -> startEiJob(eiProducer, eiJob, retrySpec)) //
  86             .collectList() //
  87             .flatMap(okResponses -> Mono.just(Integer.valueOf(okResponses.size()))); //
  88     }
  89
  90     /**
  91      * Start all jobs for one producer
  92      *
  93      * @param producer
  94      * @param eiJobs
  95      */
  96     public Flux<String> startEiJobs(EiProducer producer, EiJobs eiJobs) {
  97         final int maxNoOfParalellRequests = 10;
  98         Retry retrySpec = Retry.backoff(3, Duration.ofSeconds(1));
  99
 100         return Flux.fromIterable(producer.getEiTypes()) //
 101             .flatMap(type -> Flux.fromIterable(eiJobs.getJobsForType(type))) //
 102             .flatMap(job -> startEiJob(producer, job, retrySpec), maxNoOfParalellRequests);
 103     }
 104
 105     public Mono<String> startEiJob(EiProducer producer, EiJob eiJob, Retry retrySpec) {
 106         ProducerJobInfo request = new ProducerJobInfo(eiJob);
 107         String body = gson.toJson(request);
 108
 109         return restClient.post(producer.getJobCallbackUrl(), body) //
 110             .retryWhen(retrySpec) //
 111             .doOnNext(resp -> logger.debug("Job subscription {} started OK {}", eiJob.getId(), producer.getId())) //
 112             .onErrorResume(throwable -> {
 113                 producer.setJobDisabled(eiJob);
 114                 logger.warn("Job subscription failed {}", producer.getId(), throwable.toString());
 115                 return Mono.empty();
 116             }) //
 117             .doOnNext(resp -> producer.setJobEnabled(eiJob));
 118     }
 119
 120     private Collection<EiProducer> getProducersForJob(EiJob eiJob, EiProducers eiProducers) {
 121         return eiProducers.getProducersForType(eiJob.getTypeId());
 122     }
 123
 124 }