dmaap-adaptor-java/src/main/java/org/oran/dmaapadapter/tasks/ProducerRegstrationTask.java

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * O-RAN-SC
   4  * %%
   5  * Copyright (C) 2021 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.oran.dmaapadapter.tasks;
  22
  23 import com.google.gson.JsonParser;
  24
  25 import lombok.Getter;
  26
  27 import org.oran.dmaapadapter.clients.AsyncRestClient;
  28 import org.oran.dmaapadapter.clients.AsyncRestClientFactory;
  29 import org.oran.dmaapadapter.configuration.ApplicationConfig;
  30 import org.oran.dmaapadapter.controllers.ProducerCallbacksController;
  31 import org.oran.dmaapadapter.r1.ProducerInfoTypeInfo;
  32 import org.oran.dmaapadapter.r1.ProducerRegistrationInfo;
  33 import org.oran.dmaapadapter.repository.InfoType;
  34 import org.oran.dmaapadapter.repository.InfoTypes;
  35 import org.slf4j.Logger;
  36 import org.slf4j.LoggerFactory;
  37 import org.springframework.beans.factory.annotation.Autowired;
  38 import org.springframework.scheduling.annotation.EnableScheduling;
  39 import org.springframework.scheduling.annotation.Scheduled;
  40 import org.springframework.stereotype.Component;
  41
  42 import reactor.core.publisher.Flux;
  43 import reactor.core.publisher.Mono;
  44
  45 /**
  46  * Registers the types and this producer in ECS. This is done when needed.
  47  */
  48 @Component
  49 @EnableScheduling
  50 @SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally
  51 public class ProducerRegstrationTask {
  52
  53     private static final Logger logger = LoggerFactory.getLogger(ProducerRegstrationTask.class);
  54     private final AsyncRestClient restClient;
  55     private final ApplicationConfig applicationConfig;
  56     private final InfoTypes types;
  57     private static com.google.gson.Gson gson = new com.google.gson.GsonBuilder().create();
  58
  59     private static final String PRODUCER_ID = "DmaapGenericInfoProducer";
  60     @Getter
  61     private boolean isRegisteredInEcs = false;
  62     private static final int REGISTRATION_SUPERVISION_INTERVAL_MS = 1000 * 5;
  63
  64     public ProducerRegstrationTask(@Autowired ApplicationConfig applicationConfig, @Autowired InfoTypes types) {
  65         AsyncRestClientFactory restClientFactory = new AsyncRestClientFactory(applicationConfig.getWebClientConfig());
  66         this.restClient = restClientFactory.createRestClientNoHttpProxy("");
  67         this.applicationConfig = applicationConfig;
  68         this.types = types;
  69     }
  70
  71     @Scheduled(fixedRate = REGISTRATION_SUPERVISION_INTERVAL_MS)
  72     public void supervisionTask() {
  73         checkRegistration() //
  74                 .filter(isRegistrationOk -> !isRegistrationOk || !this.isRegisteredInEcs) //
  75                 .flatMap(isRegisterred -> registerTypesAndProducer()) //
  76                 .subscribe( //
  77                         null, //
  78                         this::handleRegistrationFailure, //
  79                         this::handleRegistrationCompleted);
  80     }
  81
  82     private void handleRegistrationCompleted() {
  83         logger.debug("Registering types and producer completed");
  84         isRegisteredInEcs = true;
  85     }
  86
  87     private void handleRegistrationFailure(Throwable t) {
  88         logger.warn("Registration of producer failed {}", t.getMessage());
  89     }
  90
  91     private Mono<Boolean> checkRegistration() {
  92         final String url = applicationConfig.getEcsBaseUrl() + "/data-producer/v1/info-producers/" + PRODUCER_ID;
  93         return restClient.get(url) //
  94                 .flatMap(this::isRegisterredInfoCorrect) //
  95                 .onErrorResume(t -> Mono.just(Boolean.FALSE));
  96     }
  97
  98     private Mono<Boolean> isRegisterredInfoCorrect(String registerredInfoStr) {
  99         ProducerRegistrationInfo registerredInfo = gson.fromJson(registerredInfoStr, ProducerRegistrationInfo.class);
 100         if (isEqual(producerRegistrationInfo(), registerredInfo)) {
 101             logger.trace("Already registered");
 102             return Mono.just(Boolean.TRUE);
 103         } else {
 104             return Mono.just(Boolean.FALSE);
 105         }
 106     }
 107
 108     private String registerTypeUrl(InfoType type) {
 109         return applicationConfig.getEcsBaseUrl() + "/data-producer/v1/info-types/" + type.getId();
 110     }
 111
 112     private Mono<String> registerTypesAndProducer() {
 113         final int CONCURRENCY = 20;
 114         final String producerUrl =
 115                 applicationConfig.getEcsBaseUrl() + "/data-producer/v1/info-producers/" + PRODUCER_ID;
 116
 117         return Flux.fromIterable(this.types.getAll()) //
 118                 .doOnNext(type -> logger.info("Registering type {}", type.getId())) //
 119                 .flatMap(type -> restClient.put(registerTypeUrl(type), gson.toJson(typeRegistrationInfo())),
 120                         CONCURRENCY) //
 121                 .collectList() //
 122                 .doOnNext(type -> logger.info("Registering producer")) //
 123                 .flatMap(resp -> restClient.put(producerUrl, gson.toJson(producerRegistrationInfo())));
 124     }
 125
 126     private Object typeSpecifcInfoObject() {
 127         return jsonObject("{}");
 128     }
 129
 130     private ProducerInfoTypeInfo typeRegistrationInfo() {
 131         return new ProducerInfoTypeInfo(jsonSchemaObject(), typeSpecifcInfoObject());
 132     }
 133
 134     private Object jsonSchemaObject() {
 135         // An object with no properties
 136         String schemaStr = "{" //
 137                 + "\"type\": \"object\"," //
 138                 + "\"properties\": {}," //
 139                 + "\"additionalProperties\": false" //
 140                 + "}"; //
 141         return jsonObject(schemaStr);
 142     }
 143
 144     private Object jsonObject(String json) {
 145         try {
 146             return JsonParser.parseString(json).getAsJsonObject();
 147         } catch (Exception e) {
 148             logger.error("Bug, error in JSON: {}", json);
 149             throw new NullPointerException(e.toString());
 150         }
 151     }
 152
 153     private boolean isEqual(ProducerRegistrationInfo a, ProducerRegistrationInfo b) {
 154         return a.jobCallbackUrl.equals(b.jobCallbackUrl) //
 155                 && a.producerSupervisionCallbackUrl.equals(b.producerSupervisionCallbackUrl) //
 156                 && a.supportedTypeIds.size() == b.supportedTypeIds.size();
 157     }
 158
 159     private ProducerRegistrationInfo producerRegistrationInfo() {
 160
 161         return ProducerRegistrationInfo.builder() //
 162                 .jobCallbackUrl(baseUrl() + ProducerCallbacksController.JOB_URL) //
 163                 .producerSupervisionCallbackUrl(baseUrl() + ProducerCallbacksController.SUPERVISION_URL) //
 164                 .supportedTypeIds(this.types.typeIds()) //
 165                 .build();
 166     }
 167
 168     private String baseUrl() {
 169         return this.applicationConfig.getSelfUrl();
 170     }
 171 }