information-coordinator-service/src/main/java/org/oransc/ics/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.ics.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.ics.clients.AsyncRestClient;
  31 import org.oransc.ics.clients.AsyncRestClientFactory;
  32 import org.oransc.ics.clients.SecurityContext;
  33 import org.oransc.ics.configuration.ApplicationConfig;
  34 import org.oransc.ics.repository.InfoJob;
  35 import org.oransc.ics.repository.InfoJobs;
  36 import org.oransc.ics.repository.InfoProducer;
  37 import org.oransc.ics.repository.InfoProducers;
  38 import org.slf4j.Logger;
  39 import org.slf4j.LoggerFactory;
  40
  41 import reactor.core.publisher.Flux;
  42 import reactor.core.publisher.Mono;
  43 import reactor.util.retry.Retry;
  44
  45 /**
  46  * Callbacks to the Producer
  47  */
  48 @SuppressWarnings("java:S3457") // No need to call "toString()" method as formatting and string ..
  49 public class ProducerCallbacks {
  50
  51     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  52     private static Gson gson = new GsonBuilder().create();
  53
  54     private final AsyncRestClient restClient;
  55
  56     public ProducerCallbacks(ApplicationConfig config, SecurityContext securityContext) {
  57         AsyncRestClientFactory restClientFactory =
  58             new AsyncRestClientFactory(config.getWebClientConfig(), securityContext);
  59         this.restClient = restClientFactory.createRestClientNoHttpProxy("");
  60     }
  61
  62     public Mono<String> healthCheck(InfoProducer producer) {
  63         return restClient.get(producer.getProducerSupervisionCallbackUrl());
  64     }
  65
  66     public void stopInfoJob(InfoJob infoJob, InfoProducers infoProducers) {
  67         for (InfoProducer producer : getProducersForJob(infoJob, infoProducers)) {
  68             String url = producer.getJobCallbackUrl() + "/" + infoJob.getId();
  69             producer.setJobDisabled(infoJob);
  70             restClient.delete(url) //
  71                 .subscribe(response -> 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      * Start a job in all producers that suports the job type
  80      *
  81      * @param infoJob an Information Job
  82      * @return the number of producers that returned OK
  83      */
  84     public Mono<Integer> startInfoSubscriptionJob(InfoJob infoJob, InfoProducers infoProducers) {
  85         Retry retrySpec = Retry.fixedDelay(1, Duration.ofSeconds(1));
  86         return Flux.fromIterable(getProducersForJob(infoJob, infoProducers)) //
  87             .flatMap(infoProducer -> startInfoJob(infoProducer, infoJob, retrySpec)) //
  88             .collectList() //
  89             .map(okResponses -> Integer.valueOf(okResponses.size())); //
  90     }
  91
  92     /**
  93      * Start all jobs for one producer
  94      *
  95      * @param producer
  96      * @param infoJobs
  97      */
  98     public Flux<String> startInfoJobs(InfoProducer producer, InfoJobs infoJobs) {
  99         final int maxNoOfParalellRequests = 10;
 100         Retry retrySpec = Retry.backoff(3, Duration.ofSeconds(1));
 101
 102         return Flux.fromIterable(producer.getInfoTypes()) //
 103             .flatMap(type -> Flux.fromIterable(infoJobs.getJobsForType(type))) //
 104             .flatMap(job -> startInfoJob(producer, job, retrySpec), maxNoOfParalellRequests);
 105     }
 106
 107     public Mono<String> startInfoJob(InfoProducer producer, InfoJob infoJob, Retry retrySpec) {
 108         ProducerJobInfo request = new ProducerJobInfo(infoJob);
 109         String body = gson.toJson(request);
 110
 111         return restClient.post(producer.getJobCallbackUrl(), body) //
 112             .retryWhen(retrySpec) //
 113             .doOnNext(resp -> logger.debug("Job subscription {} started OK {}", infoJob.getId(), producer.getId())) //
 114             .onErrorResume(throwable -> {
 115                 producer.setJobDisabled(infoJob);
 116                 logger.warn("Job subscription failed id: {} url: {}, reason: {}", producer.getId(),
 117                     producer.getJobCallbackUrl(), throwable.toString());
 118                 return Mono.empty();
 119             }) //
 120             .doOnNext(resp -> producer.setJobEnabled(infoJob));
 121     }
 122
 123     private Collection<InfoProducer> getProducersForJob(InfoJob infoJob, InfoProducers infoProducers) {
 124         return infoProducers.getProducersForType(infoJob.getTypeId());
 125     }
 126
 127 }