I trasduttori Clojure hanno lo stesso concetto delle operazioni intermedie sugli stream in Java?

Question

Mentre stavo imparando a conoscere i trasduttori in Clojure, improvvisamente mi ha colpito quello che mi hanno ricordato: Java 8 stream!

Transducers are composable algorithmic transformations. They are independent from the context of their input and output sources and specify only the essence of the transformation in terms of an individual element.

A stream is not a data structure that stores elements; instead, it conveys elements from a source such as a data structure, an array, a generator function, or an I/O channel, through a pipeline of computational operations.

Clojure:

(def xf 
    (comp 
    (filter odd?) 
    (map inc) 
    (take 5))) 

(println 
    (transduce xf + (range 100))) ; => 30 
(println 
    (into [] xf (range 100)))  ; => [2 4 6 8 10] 

Java:

// Purposely using Function and boxed primitive streams (instead of 
// UnaryOperator<LongStream>) in order to keep it general. 
Function<Stream<Long>, Stream<Long>> xf = 
     s -> s.filter(n -> n % 2L == 1L) 
       .map(n -> n + 1L) 
       .limit(5L); 

System.out.println(
     xf.apply(LongStream.range(0L, 100L).boxed()) 
       .reduce(0L, Math::addExact)); // => 30 
System.out.println(
     xf.apply(LongStream.range(0L, 100L).boxed()) 
       .collect(Collectors.toList())); // => [2, 4, 6, 8, 10] 

A parte la differenza nella digitazione statica/dinamica, questi sembrano molto simili a me per scopo e utilizzo.

L'analogia con le trasformazioni di flussi Java è un modo ragionevole di pensare ai trasduttori? In caso contrario, come è difettoso, o in che modo i due si differenziano per concetto (per non dire di implementazione)?

Answer 1

La differenza principale è che l'insieme di verbi (operazioni) è in qualche modo chiuso per i flussi mentre è aperto per trasduttori: provare per esempio per implementare partition sui flussi, si sente un po seconda classe:

import java.util.function.Function; 
import java.util.function.Supplier; 
import java.util.stream.Stream; 
import java.util.stream.Stream.Builder; 

public class StreamUtils { 
    static <T> Stream<T> delay(final Supplier<Stream<T>> thunk) { 
     return Stream.of((Object) null).flatMap(x -> thunk.get()); 
    } 

    static class Partitioner<T> implements Function<T, Stream<Stream<T>>> { 
     final Function<T, ?> f; 

     Object prev; 
     Builder<T> sb; 

     public Partitioner(Function<T, ?> f) { 
      this.f = f; 
     } 

     public Stream<Stream<T>> apply(T t) { 
      Object tag = f.apply(t); 
      if (sb != null && prev.equals(tag)) { 
       sb.accept(t); 
       return Stream.empty(); 
      } 
      Stream<Stream<T>> partition = sb == null ? Stream.empty() : Stream.of(sb.build()); 
      sb = Stream.builder(); 
      sb.accept(t); 
      prev = tag; 
      return partition; 
     } 

     Stream<Stream<T>> flush() { 
      return sb == null ? Stream.empty() : Stream.of(sb.build()); 
     } 
    } 

    static <T> Stream<Stream<T>> partitionBy(Stream<T> in, Function<T, ?> f) { 
     Partitioner<T> partitioner = new Partitioner<>(f); 
     return Stream.concat(in.flatMap(partitioner), delay(() -> partitioner.flush())); 
    } 
} 

Analogamente alle sequenze e ai riduttori, quando trasformi non crei un calcolo "più grande", crei una fonte "più grande".

Per poter eseguire calcoli, è stata introdotta la funzione xf da Streaming Stream per il sollevamento delle operazioni dai metodi alle entità di prima classe (in modo da separarli dall'origine). Così facendo hai creato un trasduttore anche se con un'interfaccia troppo grande.

Qui di seguito è una versione più generale del codice di cui sopra ad applicare qualsiasi (clojure) trasduttore ad un flusso:

import java.util.function.Function; 
import java.util.function.Supplier; 
import java.util.stream.Stream; 
import java.util.stream.Stream.Builder; 

import clojure.lang.AFn; 
import clojure.lang.IFn; 
import clojure.lang.Reduced; 

public class StreamUtils { 
    static <T> Stream<T> delay(final Supplier<Stream<T>> thunk) { 
     return Stream.of((Object) null).flatMap(x -> thunk.get()); 
    } 

    static class Transducer implements Function { 
     IFn rf; 

     public Transducer(IFn xf) { 
      rf = (IFn) xf.invoke(new AFn() { 
       public Object invoke(Object acc) { 
        return acc; 
       } 

       public Object invoke(Object acc, Object item) { 
        ((Builder<Object>) acc).accept(item); 
        return acc; 
       } 
      }); 
     } 

     public Stream<?> apply(Object t) { 
      if (rf == null) return Stream.empty(); 
      Object ret = rf.invoke(Stream.builder(), t); 
      if (ret instanceof Reduced) { 
       Reduced red = (Reduced) ret; 
       Builder<?> sb = (Builder<?>) red.deref(); 
       return Stream.concat(sb.build(), flush()); 
      } 
      return ((Builder<?>) ret).build(); 
     } 

     Stream<?> flush() { 
      if (rf == null) return Stream.empty(); 
      Builder<?> sb = (Builder<?>) rf.invoke(Stream.builder()); 
      rf = null; 
      return sb.build(); 
     } 
    } 

    static <T> Stream<?> withTransducer(Stream<T> in, IFn xf) { 
     Transducer transducer = new Transducer(xf); 
     return Stream.concat(in.flatMap(transducer), delay(() -> transducer.flush())); 
    } 
}