Provide allOf variant with access to dependent results

[zootalures]

The current version of allOf does not allow processing of results once they are available (e.g. to implement a reduce operation). It would be useful to provide a variant that allows collect and transform the dependent stage results by the user function.

To achieve this goal, here is a possible extension to FnFlow:

• New API method in com.fnproject.fn.api.flow.Flow that takes a function argument to transform the results in a lazy batch-oriented fashion:

FlowFuture<X> allResults(Flows.SerFunction<FlowResults, X> fn, FlowFuture<?>... flowFutures);

where the new interfaces FlowResult and FlowResults are defined as follows:

public interface FlowResults extends Serializable {

	/**
	 * Returns an iterator providing access to the results of one or more FlowFutures.
	 *
	 * @param maxBatchSize an upper bound on the number of results to retrieve from the completer in a
	 * single HTTP request. Results are lazily fetched from the completer during iteration and the 
	 * retrieved batch is cached in the runtime for performance.
	 *
	 * @return the iterator over the FlowFuture results
	 */
	Iterator<FlowResult> iterator(int maxBatchSize);

	/** The number of FlowFuture results contained inside this FlowResults instance.
	 * @return number of results available
	 */
	int size();
}

public interface FlowResult extends Serializable {

	/** Returns the completion value of the FlowFuture 
	 * @return the result of the given FlowFuture
	 * @throws FlowCompletionException when this future fails with an exception
	 */
	Object getValue();

	/** TBD - we could return the FlowId along with the result if we decide to 
	 * make them available via the public API
	 * @return the FlowId for the FlowFuture associated with this result
	 */
	FlowId getId();
}

• The following protocol changes would be required:
  • new operation to get a batch of FlowFuture results (using multipart in response) by FlowIds
  • allResults operation will need to serialize all dependent FlowIds as args to allResults continuation, with the runtime wrapping them inside an instance of the FlowResults interface prior to invoking the user's reduce function

[zootalures]

Snippet of original conversation from @jan-g

@gviedma suggestion (or one close to it) here:

https://github.com/fnproject/fdk-java/blob/all-results/experimental/src/main/java/com/fnproject/fn/experimental/reduce/Reduce.java

    public static class TestFn {
        public Integer handleRequest() {
            Flow fl = Flows.currentFlow();
            return Reduce.allResults(
                fl.supply(() -> 1),
                fl.supply(() -> 2),
                fl.supply(() -> 3),
                fl.supply(() -> 4),
                fl.supply(() -> 5)
            ).thenApply(results ->
                results.stream().map(f -> f.get()).reduce(0, (a, x) -> a + x)
            ).get();
        }
    }