Kava

A graph-based executor for async and sequential tasks.

// tasks are defined using the `AbstractTask` classprivateAbstractTask<Integer, Double> task1, task2, task3;
// example definition of `task2`, which is dependent upon the result of `task1`task2 = newAbstractTask<Integer, Double>() {
@OverridepublicOperatoroperator() {
// denotes whether the Task waits on all predecessor results (AND) or takes one result at a time (OR)returnOperator.AND;
}
@OverridepublicsynchronizedDoubleexecute() {
// variable to hold input valueIntegerresult;
try {
// result of the previous task is fetch using this.result(String taskName)result = this.result("task1");
// variable to hold output valueDoublemodifiedResult = result += 1.0;
Thread.sleep(1000);
returnmodifiedResult;
} catch (Exceptione) {
returnnull;
}
}
};
// each AbstractTask must be assigned a unique nametask2.name("task2");
// example definition of TaskGraphprivateTaskGraphtaskGraph = newTaskGraph();
taskGraph.link(inputTask1, someTask)
.link(inputTask2, someTask)
.link(inputTask3, someTask)
.link(someTask, outputTask)
.link(inputTask4, outputTask);
Graph.TaskGraphCallbackcompletionCallback = (ObjectfinalResult) -> System.out.println(finalResult.toString());
taskGraph.out("outputTask", completionCallback);
// data is passed to the TaskGraph using a ProcessRequest:ProcessRequest<Integer> request1 = newProcessRequest<>("data1", "inputTask1", 1);
ProcessRequest<Integer> request2 = newProcessRequest<>("data2", "inputTask2", 2);
ProcessRequest<Integer> request3 = newProcessRequest<>("data3", "inputTask3", 0);
taskGraph.in(request1);
taskGraph.in(request2);
taskGraph.in(request3);
// in the abstract execute() method of an input task (the entry point node), the data from a ProcessRequest is retrieved using `this.result(String processRequestLabel)`