@Override
  public <TInput, TOutput> VortexAggregateFuture<TInput, TOutput>
      enqueueTasklets(final VortexAggregateFunction<TOutput> aggregateFunction,
                      final VortexFunction<TInput, TOutput> vortexFunction,
                      final VortexAggregatePolicy policy,
                      final List<TInput> inputs,
                      final Optional<FutureCallback<AggregateResult<TInput, TOutput>>> callback) {
    final int aggregateFunctionId = aggregateIdCounter.getAndIncrement();
    aggregateFunctionRepository.put(aggregateFunctionId, aggregateFunction, policy);
    final List<Tasklet> tasklets = new ArrayList<>(inputs.size());
    final Map<Integer, TInput> taskletIdInputMap = new HashMap<>(inputs.size());

    for (final TInput input : inputs) {
      taskletIdInputMap.put(taskletIdCounter.getAndIncrement(), input);
    }

    final VortexAggregateFuture<TInput, TOutput> vortexAggregateFuture;
    if (callback.isPresent()) {
      vortexAggregateFuture = new VortexAggregateFuture<>(executor, taskletIdInputMap, callback.get());
    } else {
      vortexAggregateFuture = new VortexAggregateFuture<>(executor, taskletIdInputMap, null);
    }

    for (final Map.Entry<Integer, TInput> taskletIdInputEntry : taskletIdInputMap.entrySet()) {
      final Tasklet tasklet = new Tasklet<>(taskletIdInputEntry.getKey(), Optional.of(aggregateFunctionId),
          vortexFunction, taskletIdInputEntry.getValue(), vortexAggregateFuture);
      tasklets.add(tasklet);
      pendingTasklets.addLast(tasklet);
    }

    putDelegate(tasklets, vortexAggregateFuture);
    return vortexAggregateFuture;
  }
@ Override public < TInput , TOutput > VortexAggregateFuture < TInput , TOutput > enqueueTasklets ( final VortexAggregateFunction < TOutput > aggregateFunction , final VortexFunction < TInput , TOutput > vortexFunction , final VortexAggregatePolicy policy , final List < TInput > inputs , final Optional < FutureCallback < AggregateResult < TInput , TOutput > > > callback ) { final int aggregateFunctionId = aggregateIdCounter . getAndIncrement ( ) ; aggregateFunctionRepository . put ( aggregateFunctionId , aggregateFunction , policy ) ; final List < Tasklet > tasklets = new ArrayList < > ( inputs . size ( ) ) ; final Map < Integer , TInput > taskletIdInputMap = new HashMap < > ( inputs . size ( ) ) ; for ( final TInput input : inputs ) { taskletIdInputMap . put ( taskletIdCounter . getAndIncrement ( ) , input ) ; } final VortexAggregateFuture < TInput , TOutput > vortexAggregateFuture ; if ( callback . isPresent ( ) ) { vortexAggregateFuture = new VortexAggregateFuture < > ( executor , taskletIdInputMap , callback . get ( ) ) ; } else { vortexAggregateFuture = new VortexAggregateFuture < > ( executor , taskletIdInputMap , null ) ; } for ( final Map . Entry < Integer , TInput > taskletIdInputEntry : taskletIdInputMap . entrySet ( ) ) { final Tasklet tasklet = new Tasklet < > ( taskletIdInputEntry . getKey ( ) , Optional . of ( aggregateFunctionId ) , vortexFunction , taskletIdInputEntry . getValue ( ) , vortexAggregateFuture ) ; tasklets . add ( tasklet ) ; pendingTasklets . addLast ( tasklet ) ; } putDelegate ( tasklets , vortexAggregateFuture ) ; return vortexAggregateFuture ; }
Add aggregate-able Tasklets to pendingTasklets.