java.util.stream.Collectors#collectingAndThen ( )源码实例Demo

public List<CompletableFuture<Long>> executeBulk(TransactionContext txnCtx) {
    Iterable<NodeOperation> nodeOperations = nodeOperationTrees.stream()
        .flatMap(opTree -> opTree.nodeOperations().stream())
        ::iterator;
    Map<String, Collection<NodeOperation>> operationByServer = NodeOperationGrouper.groupByServer(nodeOperations);

    List<ExecutionPhase> handlerPhases = new ArrayList<>(nodeOperationTrees.size());
    List<RowConsumer> handlerConsumers = new ArrayList<>(nodeOperationTrees.size());
    List<CompletableFuture<Long>> results = new ArrayList<>(nodeOperationTrees.size());
    for (NodeOperationTree nodeOperationTree : nodeOperationTrees) {
        CollectingRowConsumer<?, Long> consumer = new CollectingRowConsumer<>(
            Collectors.collectingAndThen(Collectors.summingLong(r -> ((long) r.get(0))), sum -> sum));
        handlerConsumers.add(consumer);
        results.add(consumer.completionFuture());
        handlerPhases.add(nodeOperationTree.leaf());
    }
    try {
        setupTasks(txnCtx, operationByServer, handlerPhases, handlerConsumers);
    } catch (Throwable throwable) {
        return Collections.singletonList(CompletableFuture.failedFuture(throwable));
    }
    return results;
}
 

/**
 * Returns a {@link BiCollector} that collects the key-value pairs into an immutable {@link Map}
 * using {@code valueCollector} to collect values of identical keys into a final value of type
 * {@code V}.
 *
 * <p>For example, the following calculates total population per state from city demographic data:
 *
 * <pre>{@code
 *  Map<StateId, Integer> statePopulations = BiStream.from(cities, City::getState, c -> c)
 *     .collect(toMap(summingInt(City::getPopulation)));
 * }</pre>
 *
 * <p>Entries are collected in encounter order.
 */
public static <K, V1, V> BiCollector<K, V1, Map<K, V>> toMap(Collector<V1, ?, V> valueCollector) {
  requireNonNull(valueCollector);
  return new BiCollector<K, V1, Map<K, V>>() {
    @Override
    public <E> Collector<E, ?, Map<K, V>> splitting(
        Function<E, K> toKey, Function<E, V1> toValue) {
      return Collectors.collectingAndThen(
          Collectors.groupingBy(
              toKey,
              LinkedHashMap::new, Collectors.mapping(toValue, valueCollector)),
          Collections::unmodifiableMap);
    }
  };
}
 

public static <T> Collector<T, ?, Stream<T>> toEagerShuffledStream() {
    return Collectors.collectingAndThen(
        toCollection(ArrayList::new),
        list -> {
            Collections.shuffle(list);
            return list.stream();
        });
}
 

/**
 * Based on the query options, returns a collector that puts Datastore query results
 * in a correct form.
 *
 * @return collector
 */
private Collector<?, ?, ?> getResultsCollector() {
	Collector<?, ?, ?> collector  = Collectors.toList();
	if (isCountingQuery && !PartTreeDatastoreQuery.this.tree.isDelete()) {
		collector = Collectors.counting();
	}
	else if (PartTreeDatastoreQuery.this.tree.isExistsProjection()) {
		collector = Collectors.collectingAndThen(Collectors.counting(), (count) -> count > 0);
	}
	return collector;
}
 

/**
 * Collector of {@link IgniteInternalFuture} inheritors stream to {@link GridCompoundFuture}.
 *
 * @param <T> Result type of inheritor {@link IgniteInternalFuture}.
 * @param <R> Result type of {@link GridCompoundFuture}.
 * @return Compound future that contains all stream futures
 *         and initialized with {@link GridCompoundFuture#markInitialized()}.
 */
public static <T, R> Collector<? super IgniteInternalFuture,
    ? super GridCompoundFuture<T, R>, GridCompoundFuture<T, R>> toCompoundFuture() {
    final GridCompoundFuture<T, R> res = new GridCompoundFuture<>();

    return Collectors.collectingAndThen(
        Collectors.reducing(
            res,
            res::add,
            (a, b) -> a // No needs to merge compound futures.
        ),
        GridCompoundFuture::markInitialized
    );
}
 

public static <T> Collector<T, ?, T> singletonCollector() {
  return Collectors.collectingAndThen(
      Collectors.toList(),
      list -> {
        if (list.size() > 1) {
          throw new IllegalStateException("List contains more than one element: " + list);
        }
        return list.size() > 0 ? list.get(0) : null;
      });
}
 

private Collector<Integer, ?, LinkedList<Integer>> shufflePages() {
    return Collectors.collectingAndThen(Collectors.toCollection(LinkedList::new),
        collected -> {
            Collections.shuffle(collected);
            return collected;
        });
}
 

private static <T> Collector<T, ?, T[]> toArray(IntFunction<T[]> func) {
    return Collectors.collectingAndThen(
            Collectors.toList(), l -> l.toArray(func.apply(l.size())));
}
 

public static Collector<NodeAddress, ?, SeedNodeAddresses> collector() {
    return Collectors.collectingAndThen(Collectors.toSet(), SeedNodeAddresses::new);
}
 

public static Collector<FieldDeclaration, ?, FieldDeclarations> collector() {
    return Collectors.collectingAndThen(Collectors.toList(), FieldDeclarations::new);
}
 

public static Collector<StaticFieldDeclaration, ?, StaticFieldDeclarations> collector() {
    return Collectors.collectingAndThen(Collectors.toList(), StaticFieldDeclarations::new);
}
 

public static Collector<MethodDeclaration, ?, MethodDeclarations> collector() {
    return Collectors.collectingAndThen(toList(), MethodDeclarations::new);
}
 

public static Collector<TypeIdentifier, ?, TypeIdentifiers> collector() {
    return Collectors.collectingAndThen(Collectors.toList(), TypeIdentifiers::new);
}
 

private <T> Collector<T, ?, Set<T>> toUnmodifiableSet() {
  return Collectors.collectingAndThen(Collectors.toSet(), Collections::unmodifiableSet);
}
 

static <T> Collector<T, List<T>, LazySeq<T>> collector() {
    Collector<T, ?, List<T>> c  = Collectors.toList();
    return Collectors.<T, List<T>, Iterable<T>,LazySeq<T>>collectingAndThen((Collector)c,LazySeq::fromIterable);
}
 

/**
 * Creates a collector that collects into an unmodifiable Map.
 *
 * <p>This is a polyfill equivalent of Java 10's
 * {@code Collectors#toUnmodifiableMap}.
 *
 * @param <T> the type to retrieve keys and values from.
 * @param <K> the Map's key type.
 * @param <U> the Map's value type.
 * @param keyMapper Function that retrieves the key.
 * @param valueMapper Function that retrieves the value.
 * @return a Collector that accumulates the entries into an unmodifiable Map.
 */
@SuppressWarnings("unchecked")
public static <T, K, U> Collector<T, ?, Map<K, U>> toUnmodifiableMap(
        Function<? super T, ? extends K> keyMapper,
        Function<? super T, ? extends U> valueMapper
) {
    return Collectors.collectingAndThen(
            Collectors.toMap(
                    Objects.requireNonNull(keyMapper, "keyMapper"),
                    Objects.requireNonNull(valueMapper, "valueMapper")),
            Collections::unmodifiableMap);
}
 

/**
 * Collector used at the end of a stream to build an immutable list of
 * immutable lists of size equal to or less than size.
 * For example, the following list [a, b, c, d, e] with a partition
 * size of 2 will give [[a, b], [c, d], [e]].
 * <p>
 * A collector is used to gather data at the end of a stream operation.
 * This method returns a collector allowing streams to be gathered into
 * an {@link ImmutableList} of {@link ImmutableList}.
 *
 * @param size  the size of the partitions of the original list
 * @param <T>  the type of element in the list
 * @return the immutable list of lists collector
 */
public static <T> Collector<T, ?, ImmutableList<ImmutableList<T>>> splittingBySize(int size) {
  return Collectors.collectingAndThen(
      Collectors.collectingAndThen(
          Guavate.toImmutableList(),
          objects -> Lists.partition(objects, size)),
      Guavate::toImmutables);
}
 

static <T> Collector<T, ?, SortedSet<T>> immutableCollector() {
    return Collectors.collectingAndThen(defaultCollector(), (final SortedSet<T> d) -> Collections.unmodifiableSortedSet(d));

}
 

/**
 * Returns a collector that first copies all input elements into a new {@code Stream} and then
 * passes the stream to {@code toSink} function, which translates it to the final result.
 *
 * @since 3.6
 */
static <T, R> Collector<T, ?, R> copying(Function<Stream<T>, R> toSink) {
  return Collectors.collectingAndThen(toStream(), toSink);
}
 

/**
 * Returns a {@code Collector} which counts a number of distinct values the
 * mapper function returns for the stream elements.
 * 
 * <p>
 * The operation performed by the returned collector is equivalent to
 * {@code stream.map(mapper).distinct().count()}. This collector is mostly
 * useful as a downstream collector.
 * 
 * @param <T> the type of the input elements
 * @param mapper a function which classifies input elements.
 * @return a collector which counts a number of distinct classes the mapper
 *         function returns for the stream elements.
 * @throws NullPointerException if mapper is null.
 */
public static <T> Collector<T, ?, Integer> distinctCount(Function<? super T, ?> mapper) {
    Objects.requireNonNull(mapper);
    return Collectors.collectingAndThen(Collectors.mapping(mapper, Collectors.toSet()), Set::size);
}