ClojureトランスデューサはJavaのストリーム上の中間演算と同じ概念ですか？

Question

私はClojureのトランスデューサについて学んでいたので、突然私に思い出されたものを打ちました：Java 8ストリーム！

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] 

静的/動的型の違いとは別に、これらは目的と用途で私と全く同じように見えます。

Javaストリームの変換のアナロジーは、トランスデューサについての妥当な考え方ですか？そうでない場合、どのように欠陥があるのでしょうか？また、2つのコンセプトがどのように違うのでしょうか？

Answer 1

主な違いは、動詞（操作）のセットは何とかそれは、変換器のために開いている間のストリームのため閉鎖されていることである。

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())); 
    } 
} 

：ストリーム上のpartitionを実装するために、たとえばしようと、それはビットの第2クラスを感じていますシーケンスやレデューサーのように、 "大きな"計算を作成しないで変換すると、 "より大きな"ソースを作成します。

計算を渡すために、xfの関数をストリームからストリームに導入してメソッドからファーストクラスのエンティティに操作を持ち上げるようにしました（ソースから解くため）。そうすることで、インターフェイスが大きすぎてもトランスデューサを作成することができます。私の訓練を受けていない目に

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())); 
    } 
}