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