Prodotto cartesiano di flussi in Java 8 come flusso (utilizzando solo flussi)

Question

Vorrei creare un metodo che crea un flusso di elementi che sono prodotti cartesiani di più flussi dati (aggregati allo stesso tipo alla fine da un binario operatore). Si noti che sia gli argomenti che i risultati sono flussi, non raccolte.

Ad esempio, per due flussi di {A, B} e {X, Y} desidero che produce flusso di valori {AX, AY, BX, BY} (semplice concatenazione viene utilizzato per aggregare le stringhe). Finora, ho si avvicinò con questo codice:

private static <T> Stream<T> cartesian(BinaryOperator<T> aggregator, Stream<T>... streams) { 
    Stream<T> result = null; 

    for (Stream<T> stream : streams) { 
     if (result == null) { 
      result = stream; 
     } else { 
      result = result.flatMap(m -> stream.map(n -> aggregator.apply(m, n))); 
     } 
    } 

    return result; 
} 

Questo è il mio caso d'uso desiderato:

Stream<String> result = cartesian(
    (a, b) -> a + b, 
    Stream.of("A", "B"), 
    Stream.of("X", "Y") 
); 

System.out.println(result.collect(Collectors.toList())); 

Risultato atteso: AX, AY, BX, BY.

Un altro esempio: risultato

Stream<String> result = cartesian(
    (a, b) -> a + b, 
    Stream.of("A", "B"), 
    Stream.of("K", "L"), 
    Stream.of("X", "Y") 
); 

prevista: AKX, AKY, ALX, ALY, BKX, BKY, BLX, BLY.

Tuttavia, se corro il codice, ottengo questo errore:

IllegalStateException: flusso è già stato operati o chiuso

Dove è il flusso consumato? Per flatMap? Può essere facilmente risolto?

Answer 1

Passando i flussi nel tuo esempio non è mai meglio di liste di passaggio:

private static <T> Stream<T> cartesian(BinaryOperator<T> aggregator, List<T>... lists) { 
    ... 
} 

e usarlo in questo modo:

Stream<String> result = cartesian(
    (a, b) -> a + b, 
    Arrays.asList("A", "B"), 
    Arrays.asList("K", "L"), 
    Arrays.asList("X", "Y") 
); 

In entrambi i casi si crea un imp array lecito da varargs e usarlo come fonte di dati, quindi la pigrizia è immaginaria. I tuoi dati vengono effettivamente memorizzati negli array.

Nella maggior parte dei casi il flusso di prodotti cartesiani risultante è molto più lungo degli input, quindi non c'è praticamente alcun motivo per rendere gli input pigri. Ad esempio, con cinque elenchi di cinque elementi (25 in totale), si otterrà il flusso risultante di 3125 elementi. Quindi memorizzare 25 elementi nella memoria non è un grosso problema. In realtà nella maggior parte dei casi pratici sono già memorizzati nella memoria.

Per generare il flusso di prodotti cartesiani è necessario "riavvolgere" costantemente tutti i flussi (tranne il primo).Per riavvolgere, gli stream dovrebbero essere in grado di recuperare i dati originali ancora e ancora, o di buffering in qualche modo (cosa che non ti piace) o riprenderli di nuovo dalla sorgente (colleciton, array, file, rete, numeri casuali, ecc.) ed eseguire ancora e ancora tutte le operazioni intermedie. Se le operazioni di origine e intermedie sono lente, la soluzione lazy potrebbe essere molto più lenta della soluzione di buffering. Se la tua fonte non è in grado di produrre nuovamente i dati (ad esempio, un generatore di numeri casuali che non può produrre gli stessi numeri prodotti in precedenza), la tua soluzione sarà errata.

Tuttavia la soluzione totalmente pigra è possbile. Basta usare non corsi d'acqua, ma i fornitori Stream:

private static <T> Stream<T> cartesian(BinaryOperator<T> aggregator, 
             Supplier<Stream<T>>... streams) { 
    return Arrays.stream(streams) 
     .reduce((s1, s2) -> 
      () -> s1.get().flatMap(t1 -> s2.get().map(t2 -> aggregator.apply(t1, t2)))) 
     .orElse(Stream::empty).get(); 
} 

La soluzione è interessante che ci ha creato e ridurre il flusso dei fornitori per ottenere il fornitore risultante e, infine, lo chiamano. Uso: [? Come posso fare prodotto cartesiano con Java 8 flussi]

Stream<String> result = cartesian(
      (a, b) -> a + b, 
     () -> Stream.of("A", "B"), 
     () -> Stream.of("K", "L"), 
     () -> Stream.of("X", "Y") 
     ); 
result.forEach(System.out::println); 

Answer 2

stream si consuma nell'operazione flatMap nella seconda iterazione. Quindi devi creare un nuovo stream ogni volta che il tuo risultato è map. Pertanto è necessario raccogliere lo stream in anticipo per ottenere un nuovo flusso in ogni iterazione.

private static <T> Stream<T> cartesian(BiFunction<T, T, T> aggregator, Stream<T>... streams) { 
    Stream<T> result = null; 
    for (Stream<T> stream : streams) { 
     if (result == null) { 
      result = stream; 
     } else { 
      Collection<T> s = stream.collect(Collectors.toList()); 
      result = result.flatMap(m -> s.stream().map(n -> aggregator.apply(m, n))); 
     } 
    } 
    return result; 
} 

O ancora più breve:

private static <T> Stream<T> cartesian(BiFunction<T, T, T> aggregator, Stream<T>... streams) { 
    return Arrays.stream(streams).reduce((r, s) -> { 
     List<T> collect = s.collect(Collectors.toList()); 
     return r.flatMap(m -> collect.stream().map(n -> aggregator.apply(m, n))); 
    }).orElse(Stream.empty()); 
}