From 693026c6b973c44ade969516be496f966e4fae86 Mon Sep 17 00:00:00 2001 From: Chengkai Yan Date: Fri, 23 Nov 2018 16:23:56 +0100 Subject: fix bugs in jira DCAEGEN2-940 and DCAEGEN2-941 Change-Id: Id0b3e295cab0e085746b034caccbf82aca2e0d7b Signed-off-by: Chengkai Yan Issue-ID: DCAEGEN2-940 Issue-ID: DCAEGEN2-941 --- .../dcaegen2/collectors/datafile/tasks/ScheduledTasks.java | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) (limited to 'datafile-app-server/src/main') diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java index 171dd024..c465fe94 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java @@ -25,6 +25,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import reactor.core.publisher.Flux; +import reactor.core.scheduler.Schedulers; /** * @author Przemysław Wąsala on 3/23/18 @@ -59,11 +60,18 @@ public class ScheduledTasks { */ public void scheduleMainDatafileEventTask() { logger.trace("Execution of tasks was registered"); - + //@formatter:off consumeFromDmaapMessage() + .publishOn(Schedulers.parallel()) + .cache() .doOnError(DmaapEmptyResponseException.class, error -> logger.info("Nothing to consume from DMaaP")) - .flatMap(this::collectFilesFromXnf).flatMap(this::publishToDmaapConfiguration) + .flatMap(this::collectFilesFromXnf) + .retry(3) + .cache() + .flatMap(this::publishToDmaapConfiguration) + .retry(3) .subscribe(this::onSuccess, this::onError, this::onComplete); + //@formatter:on } private void onComplete() { -- cgit 1.2.3-korg