enrichment-coordinator-service/src/main/java/org/oransc/enrichment/controllers/r1producer/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.r1producer;
  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.InfoJob;
  34 import org.oransc.enrichment.repository.InfoJobs;
  35 import org.oransc.enrichment.repository.InfoProducer;
  36 import org.oransc.enrichment.repository.InfoProducers;
  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 Producer
  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(InfoProducer producer) {
  61         return restClient.get(producer.getProducerSupervisionCallbackUrl());
  62     }
  63
  64     public void stopInfoJob(InfoJob infoJob, InfoProducers infoProducers) {
  65         for (InfoProducer producer : getProducersForJob(infoJob, infoProducers)) {
  66             String url = producer.getJobCallbackUrl() + "/" + infoJob.getId();
  67             producer.setJobDisabled(infoJob);
  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 infoJob an Information Job
  80      * @return the number of producers that returned OK
  81      */
  82     public Mono<Integer> startInfoSubscriptionJob(InfoJob infoJob, InfoProducers infoProducers) {
  83         Retry retrySpec = Retry.fixedDelay(1, Duration.ofSeconds(1));
  84         return Flux.fromIterable(getProducersForJob(infoJob, infoProducers)) //
  85             .flatMap(infoProducer -> startInfoJob(infoProducer, infoJob, 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 infoJobs
  95      */
  96     public Flux<String> startInfoJobs(InfoProducer producer, InfoJobs infoJobs) {
  97         final int maxNoOfParalellRequests = 10;
  98         Retry retrySpec = Retry.backoff(3, Duration.ofSeconds(1));
  99
 100         return Flux.fromIterable(producer.getInfoTypes()) //
 101             .flatMap(type -> Flux.fromIterable(infoJobs.getJobsForType(type))) //
 102             .flatMap(job -> startInfoJob(producer, job, retrySpec), maxNoOfParalellRequests);
 103     }
 104
 105     public Mono<String> startInfoJob(InfoProducer producer, InfoJob infoJob, Retry retrySpec) {
 106         ProducerJobInfo request = new ProducerJobInfo(infoJob);
 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 {}", infoJob.getId(), producer.getId())) //
 112             .onErrorResume(throwable -> {
 113                 producer.setJobDisabled(infoJob);
 114                 logger.warn("Job subscription failed id: {} url: {}, reason: {}", producer.getId(),
 115                     producer.getJobCallbackUrl(), throwable.toString());
 116                 return Mono.empty();
 117             }) //
 118             .doOnNext(resp -> producer.setJobEnabled(infoJob));
 119     }
 120
 121     private Collection<InfoProducer> getProducersForJob(InfoJob infoJob, InfoProducers infoProducers) {
 122         return infoProducers.getProducersForType(infoJob.getTypeId());
 123     }
 124
 125 }