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=7b719e3c14b4f370e916e3a89e8ff75093e1cc5a;hpb=5e1623ab25b62c6c28849bfd862eba4648465922;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 7b719e3c..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 @@ -50,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 @@ -65,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) { @@ -78,7 +79,7 @@ public class ProducerRegstrationTask { @Scheduled(fixedRate = REGISTRATION_SUPERVISION_INTERVAL_MS) public void supervisionTask() { checkRegistration() // - .filter(isRegistrationOk -> !isRegistrationOk || !this.isRegisteredInEcs) // + .filter(isRegistrationOk -> !isRegistrationOk || !this.isRegisteredInIcs) // .flatMap(isRegisterred -> registerTypesAndProducer()) // .subscribe( // null, // @@ -87,8 +88,7 @@ public class ProducerRegstrationTask { } private void handleRegistrationCompleted() { - logger.debug("Registering types and producer completed"); - isRegisteredInEcs = true; + isRegisteredInIcs = true; } private void handleRegistrationFailure(Throwable t) { @@ -97,7 +97,7 @@ public class ProducerRegstrationTask { // 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)); @@ -106,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); @@ -114,13 +114,13 @@ 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())) // @@ -145,20 +145,8 @@ public class ProducerRegstrationTask { } private Object jsonSchemaObject(InfoType type) throws IOException, ServiceException { - - if (type.isKafkaTopicDefined()) { - String schemaStrKafka = readSchemaFile("/typeSchemaKafka.json"); - return jsonObject(schemaStrKafka); - } else { - // An object with no properties - String schemaStr = "{" // - + "\"type\": \"object\"," // - + "\"properties\": {}," // - + "\"additionalProperties\": false" // - + "}"; // - - return jsonObject(schemaStr); - } + String schemaFile = type.isKafkaTopicDefined() ? "/typeSchemaKafka.json" : "/typeSchemaDmaap.json"; + return jsonObject(readSchemaFile(schemaFile)); } private String readSchemaFile(String filePath) throws IOException, ServiceException { @@ -170,12 +158,13 @@ public class ProducerRegstrationTask { 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()); } } @@ -186,7 +175,6 @@ public class ProducerRegstrationTask { } private ProducerRegistrationInfo producerRegistrationInfo() { - return ProducerRegistrationInfo.builder() // .jobCallbackUrl(baseUrl() + ProducerCallbacksController.JOB_URL) // .producerSupervisionCallbackUrl(baseUrl() + ProducerCallbacksController.SUPERVISION_URL) //