diff options
Diffstat (limited to 'prh-app-server')
7 files changed, 19 insertions, 12 deletions
diff --git a/prh-app-server/config/prh_endpoints.json b/prh-app-server/config/prh_endpoints.json index 1e57e05d..e2dd51a3 100644 --- a/prh-app-server/config/prh_endpoints.json +++ b/prh-app-server/config/prh_endpoints.json @@ -4,7 +4,7 @@ "dmaapConsumerConfiguration": { "dmaapHostName": "localhost", "dmaapPortNumber": 2222, - "dmaapTopicName": "/events/unauthenticated.SEC_OTHER_OUTPUT", + "dmaapTopicName": "/events/unauthenticated.VES_PNFREG_OUTPUT", "dmaapProtocol": "http", "dmaapUserName": "admin", "dmaapUserPassword": "admin", diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/SchedulerConfig.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/SchedulerConfig.java index 8782a180..c53d3333 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/SchedulerConfig.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/SchedulerConfig.java @@ -44,8 +44,8 @@ import reactor.core.publisher.Mono; @EnableScheduling public class SchedulerConfig extends CloudConfiguration { - private static final int SCHEDULING_DELAY_FOR_PRH_TASKS = 2000; - private static final int SCHEDULING_REQUEST_FOR_CONFIGURATION_DELAY = 1; + private static final int SCHEDULING_DELAY_FOR_PRH_TASKS = 5; + private static final int SCHEDULING_REQUEST_FOR_CONFIGURATION_DELAY = 5; private static volatile List<ScheduledFuture> scheduledPrhTaskFutureList = new ArrayList<>(); private final ConcurrentTaskScheduler taskScheduler; @@ -86,7 +86,8 @@ public class SchedulerConfig extends CloudConfiguration { .scheduleAtFixedRate(super::runTask, Instant.now(), Duration.ofMinutes(SCHEDULING_REQUEST_FOR_CONFIGURATION_DELAY))); scheduledPrhTaskFutureList.add(taskScheduler - .scheduleWithFixedDelay(scheduledTask::scheduleMainPrhEventTask, SCHEDULING_DELAY_FOR_PRH_TASKS)); + .scheduleWithFixedDelay(scheduledTask::scheduleMainPrhEventTask, + Duration.ofSeconds(SCHEDULING_DELAY_FOR_PRH_TASKS))); return true; } else { return false; diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java index d7bbfd2c..53c370f1 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java @@ -29,6 +29,8 @@ import org.onap.dcaegen2.services.prh.exceptions.DmaapEmptyResponseException; import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException; import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel; import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.util.StringUtils; import reactor.core.publisher.Mono; @@ -43,6 +45,8 @@ public class DmaapConsumerJsonParser { private static final String VENDOR_NAME = "vendorName"; private static final String SERIAL_NUMBER = "serialNumber"; + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + /** * Extract info from string and create @see {@link org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel}. * @@ -51,6 +55,7 @@ public class DmaapConsumerJsonParser { */ public Mono<ConsumerDmaapModel> getJsonObject(Mono<String> monoMessage) { return monoMessage + .doOnNext(message -> logger.info("Consumed message from DmaaP: {}", message)) .flatMap(this::getJsonParserMessage) .flatMap(this::createJsonConsumerModel); } diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiProducerTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiProducerTaskImpl.java index baccd3f3..976547e2 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiProducerTaskImpl.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiProducerTaskImpl.java @@ -54,7 +54,7 @@ public class AaiProducerTaskImpl extends @Override Mono<ConsumerDmaapModel> publish(Mono<ConsumerDmaapModel> consumerDmaapModel) { - logger.info("Sending PNF model to AAI {}", consumerDmaapModel); + return aaiProducerReactiveHttpClient.getAaiProducerResponse(consumerDmaapModel) .flatMap(response -> { if (HttpUtils.isSuccessfulResponseCode(response)) { diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java index 180ad456..f8eccf11 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java @@ -57,7 +57,6 @@ public class DmaapConsumerTaskImpl extends DmaapConsumerTask { @Override Mono<ConsumerDmaapModel> consume(Mono<String> message) { - logger.info("Consumed model from DMaaP: {}", message); return dmaapConsumerJsonParser.getJsonObject(message); } diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java index 8188267e..13f1b162 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java @@ -49,9 +49,11 @@ public class DmaapPublisherTaskImpl extends DmaapPublisherTask { @Override Mono<String> publish(Mono<ConsumerDmaapModel> consumerDmaapModel) { - logger.info("Publishing on DMaaP topic {} object {}", resolveConfiguration().dmaapTopicName(), - consumerDmaapModel); - return dmaapProducerReactiveHttpClient.getDMaaPProducerResponse(consumerDmaapModel); + return consumerDmaapModel.flatMap(dmaapModel -> { + logger.info("Publishing on DMaaP topic {} object {}", resolveConfiguration().dmaapTopicName(), + dmaapModel); + return dmaapProducerReactiveHttpClient.getDMaaPProducerResponse(dmaapModel); + }); } @Override diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java index e0419e8f..662a3947 100644 --- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java +++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java @@ -92,7 +92,7 @@ class DmaapPublisherTaskImplTest { //then verify(dMaaPProducerReactiveHttpClient, times(1)) - .getDMaaPProducerResponse(any(Mono.class)); + .getDMaaPProducerResponse(any()); verifyNoMoreInteractions(dMaaPProducerReactiveHttpClient); } @@ -107,14 +107,14 @@ class DmaapPublisherTaskImplTest { .expectNext(String.valueOf(HttpStatus.UNAUTHORIZED.value())).verifyComplete(); //then - verify(dMaaPProducerReactiveHttpClient, times(1)).getDMaaPProducerResponse(any(Mono.class)); + verify(dMaaPProducerReactiveHttpClient, times(1)).getDMaaPProducerResponse(any()); verifyNoMoreInteractions(dMaaPProducerReactiveHttpClient); } private void prepareMocksForTests(Integer httpResponseCode) { dMaaPProducerReactiveHttpClient = mock(DMaaPProducerReactiveHttpClient.class); - when(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(any(Mono.class))) + when(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(any())) .thenReturn(Mono.just(httpResponseCode.toString())); dmaapPublisherTask = spy(new DmaapPublisherTaskImpl(appConfig)); when(dmaapPublisherTask.resolveConfiguration()).thenReturn(dmaapPublisherConfiguration); |