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 import java.util.Vector;
  30
  31 import org.oransc.enrichment.clients.AsyncRestClient;
  32 import org.oransc.enrichment.clients.AsyncRestClientFactory;
  33 import org.oransc.enrichment.configuration.ApplicationConfig;
  34 import org.oransc.enrichment.repository.EiJob;
  35 import org.oransc.enrichment.repository.EiJobs;
  36 import org.oransc.enrichment.repository.EiProducer;
  37 import org.oransc.enrichment.repository.EiTypes;
  38 import org.slf4j.Logger;
  39 import org.slf4j.LoggerFactory;
  40 import org.springframework.beans.factory.annotation.Autowired;
  41 import org.springframework.stereotype.Component;
  42
  43 import reactor.core.publisher.Flux;
  44 import reactor.core.publisher.Mono;
  45 import reactor.util.retry.Retry;
  46
  47 /**
  48  * Callbacks to the EiProducer
  49  */
  50 @Component
  51 @SuppressWarnings("java:S3457") // No need to call "toString()" method as formatting and string ..
  52 public class ProducerCallbacks {
  53
  54     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  55     private static Gson gson = new GsonBuilder().create();
  56
  57     private final AsyncRestClient restClient;
  58     private final EiTypes eiTypes;
  59
  60     @Autowired
  61     public ProducerCallbacks(ApplicationConfig config, EiTypes eiTypes) {
  62         AsyncRestClientFactory restClientFactory = new AsyncRestClientFactory(config.getWebClientConfig());
  63         this.restClient = restClientFactory.createRestClient("");
  64         this.eiTypes = eiTypes;
  65     }
  66
  67     public void notifyProducersJobDeleted(EiJob eiJob) {
  68         for (EiProducer producer : getProducers(eiJob)) {
  69             String url = producer.getJobCallbackUrl() + "/" + eiJob.getId();
  70             restClient.delete(url) //
  71                 .subscribe(notUsed -> logger.debug("Producer job deleted OK {}", producer.getId()), //
  72                     throwable -> logger.warn("Producer job delete failed {} {}", producer.getId(),
  73                         throwable.getMessage()),
  74                     null);
  75         }
  76     }
  77
  78     /**
  79      * Calls all producers for an EiJob activation.
  80      *
  81      * @param eiJob an EI job
  82      * @return the number of producers that returned OK
  83      */
  84     public Mono<Integer> notifyProducersJobStarted(EiJob eiJob) {
  85         Retry retrySpec = Retry.fixedDelay(1, Duration.ofSeconds(1));
  86         return Flux.fromIterable(getProducers(eiJob)) //
  87             .flatMap(eiProducer -> notifyProducerJobStarted(eiProducer, eiJob, retrySpec)) //
  88             .collectList() //
  89             .flatMap(okResponses -> Mono.just(Integer.valueOf(okResponses.size()))); //
  90     }
  91
  92     /**
  93      * Restart all jobs for one producer
  94      *
  95      * @param producer
  96      * @param eiJobs
  97      */
  98     public void restartJobs(EiProducer producer, EiJobs eiJobs) {
  99         final int maxNoOfParalellRequests = 10;
 100         Retry retrySpec = Retry.backoff(3, Duration.ofSeconds(1));
 101
 102         Flux.fromIterable(producer.getEiTypes()) //
 103             .flatMap(type -> Flux.fromIterable(eiJobs.getJobsForType(type))) //
 104             .flatMap(job -> notifyProducerJobStarted(producer, job, retrySpec), maxNoOfParalellRequests) //
 105             .onErrorResume(t -> {
 106                 logger.error("Could not restart EI Job for producer: {}, reason :{}", producer.getId(), t.getMessage());
 107                 return Flux.empty();
 108             }) //
 109             .subscribe();
 110     }
 111
 112     private Mono<String> notifyProducerJobStarted(EiProducer producer, EiJob eiJob, Retry retrySpec) {
 113         ProducerJobInfo request = new ProducerJobInfo(eiJob);
 114         String body = gson.toJson(request);
 115
 116         return restClient.post(producer.getJobCallbackUrl(), body) //
 117             .retryWhen(retrySpec) //
 118             .doOnNext(resp -> logger.debug("Job subscription {} started OK {}", eiJob.getId(), producer.getId())) //
 119             .onErrorResume(throwable -> {
 120                 logger.warn("Job subscription failed {}", producer.getId(), throwable.toString());
 121                 return Mono.empty();
 122             });
 123     }
 124
 125     private Collection<EiProducer> getProducers(EiJob eiJob) {
 126         try {
 127             return this.eiTypes.getType(eiJob.getTypeId()).getProducers();
 128         } catch (Exception e) {
 129             return new Vector<>();
 130         }
 131     }
 132
 133 }