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 void stopEiJob(EiJob eiJob, EiProducers eiProducers) {
  61         for (EiProducer producer : getProducersForJob(eiJob, eiProducers)) {
  62             String url = producer.getJobCallbackUrl() + "/" + eiJob.getId();
  63             restClient.delete(url) //
  64                 .subscribe(response -> logger.debug("Producer job deleted OK {}", producer.getId()), //
  65                     throwable -> logger.warn("Producer job delete failed {} {}", producer.getId(),
  66                         throwable.getMessage()),
  67                     null);
  68         }
  69     }
  70
  71     /**
  72      * Calls all producers for an EiJob activation.
  73      *
  74      * @param eiJob an EI job
  75      * @return the number of producers that returned OK
  76      */
  77     public Mono<Integer> startEiJob(EiJob eiJob, EiProducers eiProducers) {
  78         Retry retrySpec = Retry.fixedDelay(1, Duration.ofSeconds(1));
  79         return Flux.fromIterable(getProducersForJob(eiJob, eiProducers)) //
  80             .flatMap(eiProducer -> postStartEiJob(eiProducer, eiJob, retrySpec)) //
  81             .collectList() //
  82             .flatMap(okResponses -> Mono.just(Integer.valueOf(okResponses.size()))); //
  83     }
  84
  85     /**
  86      * Restart all jobs for one producer
  87      *
  88      * @param producer
  89      * @param eiJobs
  90      */
  91     public Flux<String> restartEiJobs(EiProducer producer, EiJobs eiJobs) {
  92         final int maxNoOfParalellRequests = 10;
  93         Retry retrySpec = Retry.backoff(3, Duration.ofSeconds(1));
  94
  95         return Flux.fromIterable(producer.getEiTypes()) //
  96             .flatMap(type -> Flux.fromIterable(eiJobs.getJobsForType(type))) //
  97             .flatMap(job -> postStartEiJob(producer, job, retrySpec), maxNoOfParalellRequests) //
  98             .onErrorResume(t -> {
  99                 logger.error("Could not restart EI Job for producer: {}, reason :{}", producer.getId(), t.getMessage());
 100                 return Flux.empty();
 101             }); //
 102
 103     }
 104
 105     private Mono<String> postStartEiJob(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             .doOnNext(resp -> producer.setJobDisabled(eiJob)) //
 113             .onErrorResume(throwable -> {
 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 }