Controlling emit values from Flux generate for sequential task execution

Question

I'm writing a simple orchestration framework using reactor framework which executes tasks sequentially, and the next task to execute is dependent on the result from previous tasks. I might have multiple paths to choose from based on the outcome of previous tasks. Earlier, I wrote a similar framework based on a static DAG where I passed as list of tasks as iterables and used Flux.fromIterable(taskList). However, this does not give me the flexibility to go dynamic because of the static array publisher.

I'm looking for alternate approaches like do(){}while(condition) to solve for DAG traversal and task decision and I came up with Flux.generate(). I evaluate the next step in generate method and pass the next task downstream. The problem I'm facing now is, Flux.generate does not wait for downstream to complete, but pushes until the condition is set to invalid. And by the time task 1 gets executed, task 2 would have been pushed n times, which is not the expected behavior.

Can someone please point me towards the right direction?

Thanks.

First iteration using List of tasks (static DAG)

Flux.fromIterable(taskList)
        .publishOn(this.factory.getSharedSchedulerPool())
        .concatMap(
            reactiveTask -> {
              log.info("Running task =>{}", reactiveTask.getTaskName());
              return reactiveTask
                  .run(ctx);
            })
        // Evaluates status from previous task and terminates stream or continues.
        .takeWhile(context -> evaluateStatus(context))
        .onErrorResume(throwable -> buildResponse(ctx, throwable))
        .doOnCancel(() -> log.info("Task cancelled"))
        .doOnComplete(() -> log.info("Completed flow"))
        .subscribe();

Attempt to dynamic dag

Flux.generate(
            (SynchronousSink> synchronousSink) -> {
              ReactiveTask task = null;
              if (ctx.getLastExecutedStep() == null) {
                // first task;
                task = getFirstTaskFromDAG();
              } else {
                task = deriveNextStep(ctx.getLastExecutedStep(), ctx.getDecisionData());
                   
              }
              if (task.getName.equals("END")) {
                synchronousSink.complete();
              }
              synchronousSink.next(task);
            })
        .publishOn(this.factory.getSharedSchedulerPool())
        .doOnNext(orchestrationContextReactiveTask -> log.info("On next => {}", 
          orchestrationContextReactiveTask.getTaskName()))
        .concatMap(
            reactiveTask -> {
              log.info("Running task =>{}", reactiveTask.getTaskName());
              return reactiveTask
                  .run(ctx);                  
            })
        .onErrorResume(throwable -> buildResponse(ctx, throwable))
        .takeUntil(context -> evaluateStatus(context, tasks))
        .doOnCancel(() -> log.info("Task cancelled"))
        .doOnComplete(() -> log.info("Completed flow")).subscribe();

The problem in above approach is, while task 1 is executing, the onNext() subscriber prints many time because generate is publishing. I want the generate method to wait on results from previous task and submit new task. In non-reactive world, this can be achieve through simple while() loop.

Each Task will perform the following action.

public class ResponseTask extends AbstractBaseTask {
  private TaskDefinition taskDefinition;
  final String taskName;


  public ResponseTask(
      StateManager stateManager,
      ThreadFactory factory,
    ) {    
    this.taskDefinition = taskDefinition;
    this.taskName = taskName;
  }

   public Mono transform(OrchestrationContext context) {
    Any masterPayload = Any.wrap(context.getIngestionPayload());
    return Mono.fromCallable(() -> stateManager.doTransformation(context, masterPayload);
  }

  
  public Mono execute(OrchestrationContext context, String payload) {
    log.info("Executing sleep for task=>{}", context.getLastExecutedStep());
    return Mono.delay(Duration.ofSeconds(1), factory.getSharedSchedulerPool())
        .then(Mono.just(context));
  }

  public Mono run(OrchestrationContext context) {
log.info("Executing task:{}. Last executed:{}", taskName, context.getLastExecutedStep());
  return transform(context)
         .doOnNext((result) -> log.info("Transformation complete for task=?{}", taskName);)
         .flatMap(payload -> {
             return execute(context, payload);
         }).onErrorResume(throwable -> {
             context.setStatus(FAILED);
             return Mono.just(context);
         }); 
}

}

EDIT - From @Ikatiforis 's recommendation - I got the following output

Here's the output from my side.

2021-12-02 09:58:14,643 INFO  (reactive_shared_pool) [ReactiveEngine lambda$doOrchestration$5:98] On next => Task1 
2021-12-02 09:58:14,644 INFO  (reactive_shared_pool) [ReactiveEngine lambda$doOrchestration$6:101] Running task =>Task1 
2021-12-02 09:58:14,644 INFO  (reactive_shared_pool) [AbstractBaseTask run:75] Executing task:Task1. Last executed:Task1 
2021-12-02 09:58:14,658 INFO  (reactive_shared_pool) [ReactiveEngine lambda$doOrchestration$5:98] On next => Task2 
2021-12-02 09:58:14,659 INFO  (reactive_shared_pool) [AbstractBaseTask lambda$run$0:83] Transformation complete for task=?Task1 
2021-12-02 09:58:14,659 INFO  (reactive_shared_pool) [ResponseTask execute:41] Executing sleep for task=>Task1 
2021-12-02 09:58:15,661 INFO  (reactive_shared_pool) [AbstractBaseTask lambda$run$4:106] Success for task=>Task1 
2021-12-02 09:58:15,663 INFO  (reactive_shared_pool) 
[ReactiveEngine lambda$doOrchestration$6:101] Running task =>Task2 
2021-12-02 09:58:15,811 INFO  (cassandra-nio-worker-8) [AbstractBaseTask run:75] Executing task:Task2. Last executed:Task2 
2021-12-02 09:58:15,811 INFO  (reactive_shared_pool) [ReactiveEngine lambda$doOrchestration$5:98] On next => Task2 
2021-12-02 09:58:15,812 INFO  (reactive_shared_pool) [AbstractBaseTask lambda$run$0:83] Transformation complete for task=?Task2 
2021-12-02 09:58:15,812 INFO  (reactive_shared_pool) [ResponseTask execute:41] Executing sleep for task=>Task2 
2021-12-02 09:58:15,837 INFO  (centaurus_reactive_shared_pool) [ReactiveEngine lambda$doOrchestration$9:113] Completed flow

I see couple of problems here --

The sequence of execution is 
 1. Task does transformations ( runs on Mono.fromCallable)
 2. Task induces a delay - Mono.fromDelay()
 3. Task completes execution. After this, generate method should evaluate the context and pass on the next task to be executed.

What I see from the output is:
 1. Task 1 starts the transformations - Runs on Mono.fromCallable.
 2. Task 2 doOnNext is reported - which means the stream already got this task.
 3. Task 1 completes.
 4. Task 2 starts and executes delay -> the stream does not wait for response from task 2 but completes the flow.

Controlling emit values from Flux generate for sequential task execution

Answers (1)

Related Questions