Copie um fluxo para evitar que "o fluxo já tenha sido operado ou fechado"

Eu gostaria de duplicar um fluxo do Java 8 para poder lidar com ele duas vezes. Eu posso collectcomo uma lista e obter novos fluxos a partir disso;

// doSomething() returns a stream
List<A> thing = doSomething().collect(toList());
thing.stream()... // do stuff
thing.stream()... // do other stuff

Mas acho que deveria haver uma maneira mais eficiente / elegante.

Existe uma maneira de copiar o fluxo sem transformá-lo em uma coleção?

Na verdade, estou trabalhando com um fluxo de Eithers, então, quero processar a projeção esquerda de uma maneira antes de passar para a projeção direita e lidar com essa outra maneira. Mais ou menos assim (com o qual até agora sou forçado a usar o toListtruque).

List<Either<Pair<A, Throwable>, A>> results = doSomething().collect(toList());

Stream<Pair<A, Throwable>> failures = results.stream().flatMap(either -> either.left());
failures.forEach(failure -> ... );

Stream<A> successes = results.stream().flatMap(either -> either.right());
successes.forEach(success -> ... );

Eu acho que sua suposição sobre eficiência é meio atrasada. Você obtém esse enorme retorno de eficiência se quiser usar os dados apenas uma vez, porque não precisa armazená-los, e os fluxos oferecem otimizações poderosas de "fusão de loop" que permitem que você flua os dados inteiros de maneira eficiente pelo pipeline.

Se você deseja reutilizar os mesmos dados, por definição, é necessário gerá-los duas vezes (deterministicamente) ou armazená-los. Se já estiver em uma coleção, ótimo; iterá-lo duas vezes é barato.

Nós experimentamos o design com "fluxos bifurcados". O que descobrimos foi que apoiar isso tinha custos reais; sobrecarregou o caso comum (use uma vez) às custas do caso incomum. O grande problema foi lidar com "o que acontece quando os dois pipelines não consomem dados na mesma taxa". Agora você voltou ao buffer de qualquer maneira. Essa era uma característica que claramente não carregava seu peso.

Se você deseja operar os mesmos dados repetidamente, armazene-os ou estruture suas operações como Consumidores e faça o seguinte:

stream()...stuff....forEach(e -> { consumerA(e); consumerB(e); });

Você também pode procurar na biblioteca RxJava, pois seu modelo de processamento se presta melhor a esse tipo de "fluxo de bifurcação".

Você pode usar uma variável local com a Supplierpara configurar partes comuns do pipeline de fluxo.

De http://winterbe.com/posts/2014/07/31/java8-stream-tutorial-examples/ :

Reutilizando fluxos

Os fluxos Java 8 não podem ser reutilizados. Assim que você chama qualquer operação do terminal, o fluxo é fechado:

Stream<String> stream = Stream.of("d2", "a2", "b1", "b3", "c")
    .filter(s -> s.startsWith("a"));
stream.anyMatch(s -> true);    // ok
stream.noneMatch(s -> true);   // exception

Calling `noneMatch` after `anyMatch` on the same stream results in the following exception:
java.lang.IllegalStateException: stream has already been operated upon or closed
at 
java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)
at 
java.util.stream.ReferencePipeline.noneMatch(ReferencePipeline.java:459)
at com.winterbe.java8.Streams5.test7(Streams5.java:38)
at com.winterbe.java8.Streams5.main(Streams5.java:28)

Para superar essa limitação, precisamos criar uma nova cadeia de fluxo para cada operação do terminal que desejamos executar; por exemplo, poderíamos criar um fornecedor de fluxo para construir um novo fluxo com todas as operações intermediárias já configuradas:

Supplier<Stream<String>> streamSupplier =
    () -> Stream.of("d2", "a2", "b1", "b3", "c")
            .filter(s -> s.startsWith("a"));

streamSupplier.get().anyMatch(s -> true);   // ok
streamSupplier.get().noneMatch(s -> true);  // ok

Cada chamada para get()constrói um novo fluxo no qual somos salvos para chamar a operação de terminal desejada.

Use a Supplierpara produzir o fluxo para cada operação de encerramento.

Supplier<Stream<Integer>> streamSupplier = () -> list.stream();

Sempre que você precisar de um fluxo dessa coleção, use streamSupplier.get()para obter um novo fluxo.

Exemplos:

1. streamSupplier.get().anyMatch(predicate);
2. streamSupplier.get().allMatch(predicate2);

Implementamos um duplicate()método para fluxos no jOOλ , uma biblioteca de código aberto que criamos para aprimorar os testes de integração para o jOOQ . Basicamente, você pode escrever:

Tuple2<Seq<A>, Seq<A>> duplicates = Seq.seq(doSomething()).duplicate();

Internamente, há um buffer armazenando todos os valores que foram consumidos de um fluxo, mas não do outro. Isso é provavelmente o mais eficiente possível, se seus dois fluxos forem consumidos aproximadamente na mesma taxa e se você puder viver com a falta de segurança de threads .

Veja como o algoritmo funciona:

static <T> Tuple2<Seq<T>, Seq<T>> duplicate(Stream<T> stream) {
    final List<T> gap = new LinkedList<>();
    final Iterator<T> it = stream.iterator();

    @SuppressWarnings("unchecked")
    final Iterator<T>[] ahead = new Iterator[] { null };

    class Duplicate implements Iterator<T> {
        @Override
        public boolean hasNext() {
            if (ahead[0] == null || ahead[0] == this)
                return it.hasNext();

            return !gap.isEmpty();
        }

        @Override
        public T next() {
            if (ahead[0] == null)
                ahead[0] = this;

            if (ahead[0] == this) {
                T value = it.next();
                gap.offer(value);
                return value;
            }

            return gap.poll();
        }
    }

    return tuple(seq(new Duplicate()), seq(new Duplicate()));
}

Mais código fonte aqui

Tuple2é, provavelmente, como o seu Pairtipo, enquanto Seqé Streamcom algumas melhorias.

Você pode criar um fluxo de executáveis ​​(por exemplo):

results.stream()
    .flatMap(either -> Stream.<Runnable> of(
            () -> failure(either.left()),
            () -> success(either.right())))
    .forEach(Runnable::run);

Onde failuree successas operações a serem aplicadas. No entanto, isso criará alguns objetos temporários e pode não ser mais eficiente do que iniciar uma coleção e transmiti-la / iterá-la duas vezes.

Outra maneira de lidar com os elementos várias vezes é usar Stream.peek (Consumer) :

doSomething().stream()
.peek(either -> handleFailure(either.left()))
.foreach(either -> handleSuccess(either.right()));

peek(Consumer) pode ser encadeado quantas vezes for necessário.

doSomething().stream()
.peek(element -> handleFoo(element.foo()))
.peek(element -> handleBar(element.bar()))
.peek(element -> handleBaz(element.baz()))
.foreach(element-> handleQux(element.qux()));

O cyclops-react , uma biblioteca na qual contribuo, possui um método estático que permite duplicar um fluxo (e retorna uma tupla de fluxos).

    Stream<Integer> stream = Stream.of(1,2,3);
    Tuple2<Stream<Integer>,Stream<Integer>> streams =  StreamUtils.duplicate(stream);

Veja os comentários, há uma penalidade de desempenho que será incorrida ao usar duplicado em um fluxo existente. Uma alternativa mais eficiente seria usar o Streamable: -

Há também uma classe Streamable (lenta) que pode ser construída a partir de um Stream, Iterable ou Array e reproduzida várias vezes.

    Streamable<Integer> streamable = Streamable.of(1,2,3);
    streamable.stream().forEach(System.out::println);
    streamable.stream().forEach(System.out::println);

AsStreamable.synchronizedFromStream (stream) - pode ser usado para criar um Streamable que preencherá lentamente sua coleção de backups, de maneira que possa ser compartilhada entre os threads. Streamable.fromStream (stream) não sofrerá nenhuma sobrecarga de sincronização.

Para esse problema específico, você também pode usar o particionamento. Algo como

     // Partition Eighters into left and right
     List<Either<Pair<A, Throwable>, A>> results = doSomething();
     Map<Boolean, Object> passingFailing = results.collect(Collectors.partitioningBy(s -> s.isLeft()));
     passingFailing.get(true) <- here will be all passing (left values)
     passingFailing.get(false) <- here will be all failing (right values)

Podemos fazer uso do Stream Builder no momento da leitura ou iteração de um fluxo. Aqui está o documento de Stream Builder .

https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.Builder.html

Caso de uso

Digamos que temos fluxo de funcionários e precisamos usá-lo para gravar dados de funcionários no arquivo do Excel e, em seguida, atualizar a coleção / tabela de funcionários [Este é apenas um caso de uso para mostrar o uso do Stream Builder]:

Stream.Builder<Employee> builder = Stream.builder();

employee.forEach( emp -> {
   //store employee data to excel file 
   // and use the same object to build the stream.
   builder.add(emp);
});

//Now this stream can be used to update the employee collection
Stream<Employee> newStream = builder.build();

Eu tive um problema semelhante e pude pensar em três estruturas intermediárias diferentes para criar uma cópia do fluxo: a List, uma matriz e a Stream.Builder. Eu escrevi um pequeno programa de benchmark, que sugeria que, do ponto de vista do desempenho, Listera cerca de 30% mais lento que os outros dois, que eram bastante semelhantes.

A única desvantagem da conversão para uma matriz é que é complicado se o seu tipo de elemento for um tipo genérico (que no meu caso era); portanto, eu prefiro usar a Stream.Builder.

Acabei escrevendo uma pequena função que cria um Collector:

private static <T> Collector<T, Stream.Builder<T>, Stream<T>> copyCollector()
{
    return Collector.of(Stream::builder, Stream.Builder::add, (b1, b2) -> {
        b2.build().forEach(b1);
        return b1;
    }, Stream.Builder::build);
}

Posso então fazer uma cópia de qualquer fluxo str, fazendo o str.collect(copyCollector())que parece bastante com o uso idiomático dos fluxos.