diff options
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() { |