Added reactive tasks flow control

Change-Id: I9cb2bede66e9e446912f2e6a815c7b56b80813b9
Issue-ID: DCAEGEN2-557
Signed-off-by: wasala <przemyslaw.wasala@nokia.com>

1 files changed, 20 insertions, 14 deletions

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 f7767101..addeaae2 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
@@ -19,11 +19,12 @@
  */
 package org.onap.dcaegen2.services.prh.tasks;
 
-import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import reactor.core.Disposable;
+import reactor.core.publisher.Mono;
 
 /**
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
@@ -46,19 +47,24 @@ public class ScheduledTasks {
     }
 
     public void scheduleMainPrhEventTask() {
-        logger.trace("Execution of task was registered");
-        setTaskExecutionFlow();
-        try {
-            dmaapConsumerTask.initConfigs();
-            dmaapConsumerTask.receiveRequest("");
-        } catch (PrhTaskException e) {
-            logger
-                .warn("Chain of tasks have been aborted, because some errors occur in prh workflow ", e);
-        }
-    }
+        logger.trace("Execution of tasks was registered");
 
-    private void setTaskExecutionFlow() {
-        dmaapConsumerTask.setNext(aaiProducerTask);
-        aaiProducerTask.setNext(dmaapProducerTask);
+        Mono.fromSupplier(() -> Mono.fromCallable(() ->
+        {
+            dmaapConsumerTask.initConfigs();
+            return dmaapConsumerTask.execute("");
+        }).subscribe(consumerDmaapModel -> Mono
+                .fromCallable(() -> aaiProducerTask.execute(consumerDmaapModel))
+                .subscribe(
+                    aaiConsumerDmaapModel -> Mono.fromCallable(() -> dmaapProducerTask.execute(aaiConsumerDmaapModel))
+                        .subscribe(resp -> logger.info("Message was published to DmaaP, response code: {}", resp),
+                            error -> logger.warn("Error has been thrown in DmaapProduerTask: {}", error),
+                            () -> logger.info("Completed DmaapPublisher task"))),
+            errorResponse -> logger
+                .warn("Error has been thrown in AAIProducerTask: {}", errorResponse)
+            , () -> logger.info("Completed AAIProducer task")))
+            .subscribe(Disposable::dispose, tasksError -> logger
+                    .warn("Chain of tasks have been aborted, because some errors occur in PRH workflow ", tasksError)
+                , () -> logger.info("PRH tasks was consumed properly")).dispose();
     }
 }