Ответ 1
Основное отличие состоит в том, что набор глаголов (операций) каким-то образом закрыт для потоков, пока он открыт для преобразователей: попробуйте, например, реализовать partition
в потоках, он чувствует себя немного вторым классом:
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()));
}
}
Также как последовательности и редукторы, когда вы трансформируете вас, вы не создаете "большее" вычисление, вы создаете "больший" источник.
Чтобы иметь возможность передавать вычисления, вы ввели xf
функцию из Stream в Stream, чтобы поднять операции от методов до объектов первого класса (чтобы развязать их из источника). Таким образом, вы создали преобразователь, хотя и со слишком большим интерфейсом.
Ниже приведена более общая версия приведенного выше кода для применения любого преобразователя (clojure) к потоку:
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()));
}
}