From 8b1502fb0f1af5d00ec26e712e57b792fbd16bd8 Mon Sep 17 00:00:00 2001 From: wasala Date: Tue, 26 Jun 2018 15:15:03 +0200 Subject: Added dmaapReactiveConsumer *Tests have not been ready yet Change-Id: I2e1d9c4218f91ae2f066b28acdbaa1870d7d27e7 Issue-ID: DCAEGEN2-557 Signed-off-by: wasala --- .../prh/service/DmaapConsumerJsonParser.java | 40 ++++++++++++++-------- .../services/prh/tasks/DmaapConsumerTask.java | 9 ++--- .../services/prh/tasks/DmaapConsumerTaskImpl.java | 30 ++++++++-------- .../services/prh/tasks/ScheduledTasks.java | 36 ++++++++++--------- 4 files changed, 66 insertions(+), 49 deletions(-) (limited to 'prh-app-server/src/main/java/org/onap') 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 ee42ce4a..20ec78fc 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 @@ -26,11 +26,12 @@ import java.util.Optional; import java.util.stream.StreamSupport; import org.onap.dcaegen2.services.prh.exceptions.DmaapEmptyResponseException; import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException; -import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException; 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; /** * @author Przemysław Wąsala on 5/8/18 @@ -46,19 +47,29 @@ public class DmaapConsumerJsonParser { private static final String PNF_SERIAL_NUMBER = "pnfSerialNumber"; - public Optional getJsonObject(String message) throws PrhTaskException { - JsonElement jsonElement = new JsonParser().parse(message); - Optional consumerDmaapModel; - if (jsonElement.isJsonObject()) { - consumerDmaapModel = Optional.of(create(jsonElement.getAsJsonObject())); - } else { - consumerDmaapModel = Optional - .of(create(StreamSupport.stream(jsonElement.getAsJsonArray().spliterator(), false).findFirst() - .flatMap(this::getJsonObjectFromAnArray) - .orElseThrow(DmaapEmptyResponseException::new))); - } - logger.info("Parsed model from DmaaP after getting it: {}", consumerDmaapModel); - return consumerDmaapModel; + public Mono getJsonObject(Mono monoMessage) { + return monoMessage.flatMap(message -> + { + if (!StringUtils.isEmpty(message)) { + JsonElement jsonElement = new JsonParser().parse(message); + ConsumerDmaapModel consumerDmaapModel; + try { + if (jsonElement.isJsonObject()) { + consumerDmaapModel = create(jsonElement.getAsJsonObject()); + } else { + consumerDmaapModel = create( + StreamSupport.stream(jsonElement.getAsJsonArray().spliterator(), false).findFirst() + .flatMap(this::getJsonObjectFromAnArray) + .orElseThrow(DmaapEmptyResponseException::new)); + } + logger.info("Parsed model from DmaaP after getting it: {}", consumerDmaapModel); + return Mono.just(consumerDmaapModel); + } catch (DmaapNotFoundException | DmaapEmptyResponseException e) { + return Mono.error(e); + } + } + return Mono.error(new DmaapEmptyResponseException()); + }); } public Optional getJsonObjectFromAnArray(JsonElement element) { @@ -101,5 +112,4 @@ public class DmaapConsumerJsonParser { private boolean containsHeader(JsonObject jsonObject) { return jsonObject.has(EVENT) && jsonObject.getAsJsonObject(EVENT).has(OTHER_FIELDS); } - } diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java index 1be3b28d..d238b34c 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java @@ -21,18 +21,19 @@ package org.onap.dcaegen2.services.prh.tasks; import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException; import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel; -import org.onap.dcaegen2.services.prh.service.consumer.ExtendedDmaapConsumerHttpClientImpl; +import org.onap.dcaegen2.services.prh.service.consumer.DmaapConsumerReactiveHttpClient; +import reactor.core.publisher.Mono; /** * @author Przemysław Wąsala on 4/13/18 */ abstract class DmaapConsumerTask { - abstract ConsumerDmaapModel consume(String message) throws PrhTaskException; + abstract Mono consume(Mono message) throws PrhTaskException; - abstract ExtendedDmaapConsumerHttpClientImpl resolveClient(); + abstract DmaapConsumerReactiveHttpClient resolveClient(); abstract void initConfigs(); - protected abstract ConsumerDmaapModel execute(String object) throws PrhTaskException; + protected abstract Mono execute(String object) throws PrhTaskException; } 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 3944d416..564a7a41 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 @@ -23,15 +23,17 @@ import java.util.Optional; import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration; import org.onap.dcaegen2.services.prh.configuration.AppConfig; import org.onap.dcaegen2.services.prh.configuration.Config; +import org.onap.dcaegen2.services.prh.exceptions.DmaapEmptyResponseException; import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException; import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException; import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel; import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser; -import org.onap.dcaegen2.services.prh.service.consumer.ExtendedDmaapConsumerHttpClientImpl; +import org.onap.dcaegen2.services.prh.service.consumer.DmaapConsumerReactiveHttpClient; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; +import reactor.core.publisher.Mono; /** * @author Przemysław Wąsala on 3/23/18 @@ -41,8 +43,8 @@ public class DmaapConsumerTaskImpl extends DmaapConsumerTask { private final Logger logger = LoggerFactory.getLogger(this.getClass()); private final Config prhAppConfig; - private ExtendedDmaapConsumerHttpClientImpl extendedDmaapConsumerHttpClient; private DmaapConsumerJsonParser dmaapConsumerJsonParser; + private DmaapConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient; @Autowired public DmaapConsumerTaskImpl(AppConfig prhAppConfig) { @@ -57,18 +59,18 @@ public class DmaapConsumerTaskImpl extends DmaapConsumerTask { @Override - ConsumerDmaapModel consume(String message) throws PrhTaskException { - logger.info("Consumed model from DMaaP: {}", message); - return dmaapConsumerJsonParser.getJsonObject(message) - .orElseThrow(() -> new DmaapNotFoundException("Null response from JSON Object in single request")); + Mono consume(Mono message) { + logger.info("Consumed model from DmaaP: {}", message); + return dmaapConsumerJsonParser.getJsonObject(message); } + @Override - public ConsumerDmaapModel execute(String object) throws PrhTaskException { - extendedDmaapConsumerHttpClient = resolveClient(); + public Mono execute(String object) { + dmaapConsumerReactiveHttpClient = resolveClient(); +// dmaapConsumerReactiveHttpClient.initWebClient(); logger.trace("Method called with arg {}", object); - return consume((extendedDmaapConsumerHttpClient.getHttpConsumerResponse().orElseThrow(() -> - new PrhTaskException("DMaaPConsumerTask has returned null")))); + return consume((dmaapConsumerReactiveHttpClient.getDmaaPConsumerResposne())); } @Override @@ -81,8 +83,8 @@ public class DmaapConsumerTaskImpl extends DmaapConsumerTask { } @Override - ExtendedDmaapConsumerHttpClientImpl resolveClient() { - return Optional.ofNullable(extendedDmaapConsumerHttpClient) - .orElseGet(() -> new ExtendedDmaapConsumerHttpClientImpl(resolveConfiguration())); + DmaapConsumerReactiveHttpClient resolveClient() { + return Optional.ofNullable(dmaapConsumerReactiveHttpClient) + .orElseGet(() -> new DmaapConsumerReactiveHttpClient(resolveConfiguration())); } -} \ No newline at end of file +} diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java index cf096b7b..37b8686e 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java @@ -55,7 +55,7 @@ public class ScheduledTasks { Mono dmaapProducerResponse = Mono.fromCallable(consumeFromDMaaPMessage()) .doOnError(DmaapEmptyResponseException.class, error -> logger.warn("Nothing to consume from DMaaP")) - .flatMap(this::publishToAAIConfiguration) + .map(this::publishToAAIConfiguration) .flatMap(this::publishToDMaaPConfiguration) .subscribeOn(Schedulers.elastic()); @@ -76,7 +76,7 @@ public class ScheduledTasks { } } - private Callable consumeFromDMaaPMessage() { + private Callable> consumeFromDMaaPMessage() { return () -> { dmaapConsumerTask.initConfigs(); @@ -84,21 +84,25 @@ public class ScheduledTasks { }; } - private Mono publishToAAIConfiguration(ConsumerDmaapModel dmaapModel) { - try { - return Mono.just(aaiProducerTask.execute(dmaapModel)); - } catch (PrhTaskException e) { - logger.warn("Exception in A&AIProducer task ", e); - return Mono.error(e); - } + private Mono publishToAAIConfiguration(Mono monoDMaaPModel) { + return monoDMaaPModel.flatMap(dmaapModel -> { + try { + return Mono.just(aaiProducerTask.execute(dmaapModel)); + } catch (PrhTaskException e) { + logger.warn("Exception in A&AIProducer task ", e); + return Mono.error(e); + } + }); } - private Mono publishToDMaaPConfiguration(ConsumerDmaapModel aaiModel) { - try { - return Mono.just(dmaapProducerTask.execute(aaiModel)); - } catch (PrhTaskException e) { - logger.warn("Exception in DMaaPProducer task ", e); - return Mono.error(e); - } + private Mono publishToDMaaPConfiguration(Mono monoAAIModel) { + return monoAAIModel.flatMap(aaiModel -> { + try { + return Mono.just(dmaapProducerTask.execute(aaiModel)); + } catch (PrhTaskException e) { + logger.warn("Exception in DMaaPProducer task ", e); + return Mono.error(e); + } + }); } } -- cgit 1.2.3-korg