X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=dmaap-adaptor-java%2Fsrc%2Fmain%2Fjava%2Forg%2Foran%2Fdmaapadapter%2Ftasks%2FProducerRegstrationTask.java;h=c9284b5bbd03836700ab882b1d564206c272cf14;hb=86f81813c94a44337c199124e7bbf6280e2c6aa6;hp=837ca323d95b0ee44e7ccb0406cbdf0cea32ec9e;hpb=54e5f0a956ea4a9f5a31b3b6319f8f3b2b980c91;p=nonrtric.git diff --git a/dmaap-adaptor-java/src/main/java/org/oran/dmaapadapter/tasks/ProducerRegstrationTask.java b/dmaap-adaptor-java/src/main/java/org/oran/dmaapadapter/tasks/ProducerRegstrationTask.java index 837ca323..c9284b5b 100644 --- a/dmaap-adaptor-java/src/main/java/org/oran/dmaapadapter/tasks/ProducerRegstrationTask.java +++ b/dmaap-adaptor-java/src/main/java/org/oran/dmaapadapter/tasks/ProducerRegstrationTask.java @@ -20,14 +20,21 @@ package org.oran.dmaapadapter.tasks; +import com.google.common.io.CharStreams; import com.google.gson.JsonParser; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; + import lombok.Getter; import org.oran.dmaapadapter.clients.AsyncRestClient; import org.oran.dmaapadapter.clients.AsyncRestClientFactory; import org.oran.dmaapadapter.configuration.ApplicationConfig; import org.oran.dmaapadapter.controllers.ProducerCallbacksController; +import org.oran.dmaapadapter.exceptions.ServiceException; import org.oran.dmaapadapter.r1.ProducerInfoTypeInfo; import org.oran.dmaapadapter.r1.ProducerRegistrationInfo; import org.oran.dmaapadapter.repository.InfoType; @@ -43,7 +50,8 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; /** - * Registers the types and this producer in ECS. This is done when needed. + * Registers the types and this producer in Innformation Coordinator Service. + * This is done when needed. */ @Component @EnableScheduling @@ -58,7 +66,7 @@ public class ProducerRegstrationTask { private static final String PRODUCER_ID = "DmaapGenericInfoProducer"; @Getter - private boolean isRegisteredInEcs = false; + private boolean isRegisteredInIcs = false; private static final int REGISTRATION_SUPERVISION_INTERVAL_MS = 1000 * 5; public ProducerRegstrationTask(@Autowired ApplicationConfig applicationConfig, @Autowired InfoTypes types) { @@ -71,7 +79,7 @@ public class ProducerRegstrationTask { @Scheduled(fixedRate = REGISTRATION_SUPERVISION_INTERVAL_MS) public void supervisionTask() { checkRegistration() // - .filter(isRegisterred -> !isRegisterred) // + .filter(isRegistrationOk -> !isRegistrationOk || !this.isRegisteredInIcs) // .flatMap(isRegisterred -> registerTypesAndProducer()) // .subscribe( // null, // @@ -80,17 +88,16 @@ public class ProducerRegstrationTask { } private void handleRegistrationCompleted() { - logger.debug("Registering types and producer succeeded"); - isRegisteredInEcs = true; + isRegisteredInIcs = true; } private void handleRegistrationFailure(Throwable t) { - logger.warn("Registration failed {}", t.getMessage()); - isRegisteredInEcs = false; + logger.warn("Registration of producer failed {}", t.getMessage()); } + // Returns TRUE if registration is correct private Mono checkRegistration() { - final String url = applicationConfig.getEcsBaseUrl() + "/data-producer/v1/info-producers/" + PRODUCER_ID; + final String url = applicationConfig.getIcsBaseUrl() + "/data-producer/v1/info-producers/" + PRODUCER_ID; return restClient.get(url) // .flatMap(this::isRegisterredInfoCorrect) // .onErrorResume(t -> Mono.just(Boolean.FALSE)); @@ -99,7 +106,7 @@ public class ProducerRegstrationTask { private Mono isRegisterredInfoCorrect(String registerredInfoStr) { ProducerRegistrationInfo registerredInfo = gson.fromJson(registerredInfoStr, ProducerRegistrationInfo.class); if (isEqual(producerRegistrationInfo(), registerredInfo)) { - logger.trace("Already registered"); + logger.trace("Already registered in ICS"); return Mono.just(Boolean.TRUE); } else { return Mono.just(Boolean.FALSE); @@ -107,17 +114,17 @@ public class ProducerRegstrationTask { } private String registerTypeUrl(InfoType type) { - return applicationConfig.getEcsBaseUrl() + "/data-producer/v1/info-types/" + type.getId(); + return applicationConfig.getIcsBaseUrl() + "/data-producer/v1/info-types/" + type.getId(); } private Mono registerTypesAndProducer() { final int CONCURRENCY = 20; final String producerUrl = - applicationConfig.getEcsBaseUrl() + "/data-producer/v1/info-producers/" + PRODUCER_ID; + applicationConfig.getIcsBaseUrl() + "/data-producer/v1/info-producers/" + PRODUCER_ID; return Flux.fromIterable(this.types.getAll()) // .doOnNext(type -> logger.info("Registering type {}", type.getId())) // - .flatMap(type -> restClient.put(registerTypeUrl(type), gson.toJson(typeRegistrationInfo())), + .flatMap(type -> restClient.put(registerTypeUrl(type), gson.toJson(typeRegistrationInfo(type))), CONCURRENCY) // .collectList() // .doOnNext(type -> logger.info("Registering producer")) // @@ -128,26 +135,36 @@ public class ProducerRegstrationTask { return jsonObject("{}"); } - private ProducerInfoTypeInfo typeRegistrationInfo() { - return new ProducerInfoTypeInfo(jsonSchemaObject(), typeSpecifcInfoObject()); + private ProducerInfoTypeInfo typeRegistrationInfo(InfoType type) { + try { + return new ProducerInfoTypeInfo(jsonSchemaObject(type), typeSpecifcInfoObject()); + } catch (Exception e) { + logger.error("Fatal error {}", e.getMessage()); + return null; + } + } + + private Object jsonSchemaObject(InfoType type) throws IOException, ServiceException { + String schemaFile = type.isKafkaTopicDefined() ? "/typeSchemaKafka.json" : "/typeSchemaDmaap.json"; + return jsonObject(readSchemaFile(schemaFile)); } - private Object jsonSchemaObject() { - // An object with no properties - String schemaStr = "{" // - + "\"type\": \"object\"," // - + "\"properties\": {}," // - + "\"additionalProperties\": false" // - + "}"; // - return jsonObject(schemaStr); + private String readSchemaFile(String filePath) throws IOException, ServiceException { + InputStream in = getClass().getResourceAsStream(filePath); + logger.debug("Reading application schema file from: {} with: {}", filePath, in); + if (in == null) { + throw new ServiceException("Could not readfile: " + filePath); + } + return CharStreams.toString(new InputStreamReader(in, StandardCharsets.UTF_8)); } + @SuppressWarnings("java:S2139") // Log exception private Object jsonObject(String json) { try { return JsonParser.parseString(json).getAsJsonObject(); } catch (Exception e) { - logger.error("Bug, error in JSON: {}", json); - throw new NullPointerException(e.toString()); + logger.error("Bug, error in JSON: {} {}", json, e.getMessage()); + throw new NullPointerException(e.getMessage()); } } @@ -158,7 +175,6 @@ public class ProducerRegstrationTask { } private ProducerRegistrationInfo producerRegistrationInfo() { - return ProducerRegistrationInfo.builder() // .jobCallbackUrl(baseUrl() + ProducerCallbacksController.JOB_URL) // .producerSupervisionCallbackUrl(baseUrl() + ProducerCallbacksController.SUPERVISION_URL) //