Java 8 génériques: Réduire un flux de consommateurs à un seul consommateur

Question

Comment puis-je écrire une méthode pour combiner une Stream de Consumers en une seule, Consumer l'aide Consumer.andThen(Consumer)?

Ma première version était:

<T> Consumer<T> combine(Stream<Consumer<T>> consumers) {
    return consumers
            .filter(Objects::nonNull)
            .reduce(Consumer::andThen)
            .orElse(noOpConsumer());
}

<T> Consumer<T> noOpConsumer() {
    return value -> { /* do nothing */ };
}

Cette version compile avec JavaC et Eclipse. Mais il est trop spécifique: L' Stream ne peut pas être un Stream<SpecialConsumer>, et si l' Consumers ne sont pas exactement du type T mais un super type, il ne peut pas être utilisé:

Stream<? extends Consumer<? super Foo>> consumers = ... ;
combine(consumers);

Qui ne compile pas, à juste titre. La version améliorée serait:

<T> Consumer<T> combine(Stream<? extends Consumer<? super T>> consumers) {
    return consumers
            .filter(Objects::nonNull)
            .reduce(Consumer::andThen)
            .orElse(noOpConsumer());
}

Mais ni l'Éclipse, ni compilation JavaC que:
Eclipse (4.7.3 un):

Le type Consumer ne définit pas andThen(capture#7-of ? extends Consumer<? super T>, capture#7-of ? extends Consumer<? super T>) qui est applicable ici

JavaC (1.8.0172):

error: incompatible types: invalid méthode de référence
.reduce(Consumer::andThen)
des types incompatibles: Consumer<CAP#1> ne peut pas être convertie Consumer<? super CAP#2>
où T est un type de variable:
T extends Object déclaré dans la méthode <T>combine(Stream<? extends Consumer<? super T>>)
où CAP#1,CAP#2 sont neuves par type de variables:
CAP#1 extends Object super: T from capture of ? super T
CAP#2 extends Object super: T from capture of ? super T

Mais cela devrait fonctionner: Chaque sous-classe de Consommateur peut être utilisé en tant que Consommateur, trop. Et tous les Consommateurs d'un super-type de X peut consommer Xs, trop. J'ai essayé d'ajouter des paramètres de type pour chaque ligne du flux version, mais ça n'aide pas. Mais si je l'écris avec un traditionnel de la boucle, il compile:

<T> Consumer<T> combine(Collection<? extends Consumer<? super T>> consumers) {
    Consumer<T> result = noOpConsumer()
    for (Consumer<? super T> consumer : consumers) {
        result = result.andThen(consumer);
    }
    return result;
}

(Le filtrage des valeurs null est laissée de côté pour alléger le texte.)

Donc, ma question est: Comment puis-je convaincre JavaC et Eclipse que mon Code est correct? Ou, si il n'est pas correct: Pourquoi la boucle-version correcte, mais pas l' Stream Version?

Answer 1

Vous utilisez un argument Stream.reduce(accumulator) version qui a la signature suivante:

Optional<T> reduce(BinaryOperator<T> accumulator);

L' BinaryOperator<T> accumulator ne peuvent accepter que des éléments de type T, mais vous devez:

<? extends Consumer<? super T>>

Je vous propose d'utiliser un trois-argument de la version de l' Stream.reduce(...) méthode de:

<U> U reduce(U identity,
             BiFunction<U, ? super T, U> accumulator
             BinaryOperator<U> combiner);

L' BiFunction<U, ? super T, U> accumulator pouvez accepter les paramètres de deux types différents, a moins restrictif lié et est plus approprié pour votre situation. Une solution possible pourrait être:

<T> Consumer<T> combine(Stream<? extends Consumer<? super T>> consumers) {
    return consumers.filter(Objects::nonNull)
                    .reduce(t -> {}, Consumer::andThen, Consumer::andThen);
}

Le troisième argument BinaryOperator<U> combiner est appelée uniquement dans le flux parallèles, mais de toute façon, il serait sage de fournir une correcte mise en œuvre.

En outre, pour une meilleure compréhension, on pourrait représenter le code ci-dessus comme suit:

<T> Consumer<T> combine(Stream<? extends Consumer<? super T>> consumers) {

    Consumer<T> identity = t -> {};
    BiFunction<Consumer<T>, Consumer<? super T>, Consumer<T>> acc = Consumer::andThen;
    BinaryOperator<Consumer<T>> combiner = Consumer::andThen;

    return consumers.filter(Objects::nonNull)
                    .reduce(identity, acc, combiner);
}

Maintenant, vous pouvez écrire:

Stream<? extends Consumer<? super Foo>> consumers = Stream.of();
combine(consumers);

Answer 2

Vous avez oublié une petite chose dans votre définition de la méthode. Il est actuellement:

<T> Consumer<T> combine(Stream<? extends Consumer<? super T>> consumers) {}

Mais vous êtes returing Consumer<? super T>. Donc, en changeant le type de retour, il fonctionne presque. Maintenant, vous accepter un argument consumers de type Stream<? extends Consumer<? super T>>. Actuellement, il ne fonctionne pas, parce que vous travaillez avec éventuellement des différentes sous-classes et la mise en œuvre de l' Consumer<? super T> (en raison de la upperbounded générique extends). Vous pouvez surmonter ce, par la coulée de chaque ? extends Consumer<? super T> votre Stream d'un simple Consumer<? super T>. Comme suit:

<T> Consumer<? super T> combine(Stream<? extends Consumer<? super T>> consumers) {
    return consumers
        .filter(Objects::nonNull)
        .map(c -> (Consumer<? super T>) c)
        .reduce(Consumer::andThen)
        .orElse(noOpConsumer());
}

Cela devrait maintenant fonctionner

Answer 3

Si vous avez beaucoup de consommateurs, en appliquant Consumer.andThen() va créer un énorme arbre de la consommation des wrappers qui est traitée de manière récursive à l'appel original de chaque consommateur.

Il pourrait donc être plus efficace de créer simplement une liste des consommateurs, et de créer un simple consommateur qui effectue une itération sur eux:

<T> Consumer<T> combine(Stream<? extends Consumer<? super T>> consumers) {
    List<Consumer<? super T>> consumerList = consumers
            .filter(Objects::nonNull)
            .collect(Collectors.toList());
    return t -> consumerList.forEach(c -> c.accept(t));
}

Alternativement, si vous pouvez garantir que les consommateurs ne sera appelée qu'une fois, et que l' Stream sera encore valide, vous pouvez simplement effectuer une itération directement sur le flux:

return t -> consumers
        .filter(Objects::nonNull)
        .forEach(c -> c.accept(t));