diff options
| author |   Chengkai Yan <martin.c.yan@est.tech> | 2018-11-23 16:23:56 +0100 |
|---|---|---|
| committer |   elinuxhenrik <henrik.b.andersson@est.tech> | 2018-12-03 13:20:42 +0100 |
| commit | 693026c6b973c44ade969516be496f966e4fae86 (patch) | |
| tree | 8398e3e98d79303d5d53c51eeccf3db018e2865b /datafile-app-server/src/main | |
| parent | a3c452af58c12283d76019509dd605f67f14532c (diff) | |
fix bugs in jira DCAEGEN2-940 and DCAEGEN2-941
Change-Id: Id0b3e295cab0e085746b034caccbf82aca2e0d7b
Signed-off-by: Chengkai Yan <martin.c.yan@est.tech>
Issue-ID: DCAEGEN2-940
Issue-ID: DCAEGEN2-941
Diffstat (limited to 'datafile-app-server/src/main')
| -rw-r--r-- | datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java | 12 |
1 files changed, 10 insertions, 2 deletions
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 <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> 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() { |