Разделить поток по функции дискриминатора
Одной из недостающих функций API Streams является преобразование "разделение", например, как определено в Clojure. Скажем, я хочу воспроизвести соединение Hibernate fetch: я хочу выпустить одну инструкцию SQL SELECT для получения таких объектов из результата:
class Family {
String surname;
List<String> members;
}
Я выпускаю:
SELECT f.name, m.name
FROM Family f JOIN Member m on m.family_id = f.id
ORDER BY f.name
и я получаю плоский поток записей (f.name, m.name)
. Теперь мне нужно преобразовать его в поток объектов Family
, со списком его членов внутри. Предположим, что у меня уже есть Stream<ResultRow>
; теперь мне нужно преобразовать его в Stream<List<ResultRow>>
, а затем действовать на него с преобразованием отображения, которое превращает его в Stream<Family>
.
Семантика преобразования заключается в следующем: продолжайте собирать поток в List
до тех пор, пока предоставленная функция дискриминатора продолжает возвращать одно и то же значение; как только значение изменится, испустите List
как элемент выходного потока и начните собирать новый List
.
Я надеюсь, что смогу написать такой код (у меня уже есть метод resultStream
):
Stream<ResultRow> dbStream = resultStream(queryBuilder.createQuery(
"SELECT f.name, m.name"
+ " FROM Family f JOIN Member m on m.family_id = f.id"
+ " ORDER BY f.name"));
Stream<List<ResultRow> partitioned = partitionBy(r -> r.string(0), dbStream);
Stream<Family> = partitioned.map(rs -> {
Family f = new Family(rs.get(0).string(0));
f.members = rs.stream().map(r -> r.string(1)).collect(toList());
return f;
});
Разумеется, я ожидаю, что результирующий поток останется ленивым (не материализованным), поскольку я хочу иметь возможность обрабатывать результирующий набор любого размера, не затрагивая никаких ограничений памяти O (n). Без этого важного требования я был бы доволен предоставленным коллекционером groupingBy
.
Ответы
Ответ 1
Решение требует от нас определения пользовательского Spliterator
, который можно использовать для построения секционированного потока. Нам нужно будет получить доступ к входному потоку через свой собственный разделитель и обернуть его в наш. Выходной поток затем создается из нашего пользовательского разделителя.
Следующий Spliterator превратит любой Stream<E>
в Stream<List<E>>
, если a Function<E, ?>
в качестве функции дискриминатора. Обратите внимание, что входной поток должен быть упорядочен для этой операции.
public class PartitionBySpliterator<E> extends AbstractSpliterator<List<E>> {
private final Spliterator<E> spliterator;
private final Function<? super E, ?> partitionBy;
private HoldingConsumer<E> holder;
private Comparator<List<E>> comparator;
public PartitionBySpliterator(Spliterator<E> toWrap, Function<? super E, ?> partitionBy) {
super(Long.MAX_VALUE, toWrap.characteristics() & ~SIZED | NONNULL);
this.spliterator = toWrap;
this.partitionBy = partitionBy;
}
public static <E> Stream<List<E>> partitionBy(Function<E, ?> partitionBy, Stream<E> in) {
return StreamSupport.stream(new PartitionBySpliterator<>(in.spliterator(), partitionBy), false);
}
@Override public boolean tryAdvance(Consumer<? super List<E>> action) {
final HoldingConsumer<E> h;
if (holder == null) {
h = new HoldingConsumer<>();
if (!spliterator.tryAdvance(h)) return false;
holder = h;
}
else h = holder;
final ArrayList<E> partition = new ArrayList<>();
final Object partitionKey = partitionBy.apply(h.value);
boolean didAdvance;
do partition.add(h.value);
while ((didAdvance = spliterator.tryAdvance(h))
&& Objects.equals(partitionBy.apply(h.value), partitionKey));
if (!didAdvance) holder = null;
action.accept(partition);
return true;
}
static final class HoldingConsumer<T> implements Consumer<T> {
T value;
@Override public void accept(T value) { this.value = value; }
}
@Override public Comparator<? super List<E>> getComparator() {
final Comparator<List<E>> c = this.comparator;
return c != null? c : (this.comparator = comparator());
}
private Comparator<List<E>> comparator() {
@SuppressWarnings({"unchecked","rawtypes"})
final Comparator<? super E> innerComparator =
Optional.ofNullable(spliterator.getComparator())
.orElse((Comparator) naturalOrder());
return (left, right) -> {
final int c = innerComparator.compare(left.get(0), right.get(0));
return c != 0? c : innerComparator.compare(
left.get(left.size() - 1), right.get(right.size() - 1));
};
}
}
Ответ 2
Для тех из вас, кто просто хочет разделить поток, для этого есть картографы и коллекторы.
class Person {
String surname;
String forename;
public Person(String surname, String forename) {
this.surname = surname;
this.forename = forename;
}
@Override
public String toString() {
return forename;
}
}
class Family {
String surname;
List<Person> members;
public Family(String surname, List<Person> members) {
this.surname = surname;
this.members = members;
}
@Override
public String toString() {
return "Family{" + "surname=" + surname + ", members=" + members + '}';
}
}
private void test() {
String[][] data = {
{"Kray", "Ronald"},
{"Kray", "Reginald"},
{"Dors", "Diana"},};
// Their families.
Stream<Family> families = Arrays.stream(data)
// Build people
.map(a -> new Person(a[0], a[1]))
// Collect into a Map<String,List<Person>> as families
.collect(Collectors.groupingBy(p -> p.surname))
// Convert them to families.
.entrySet().stream()
.map(p -> new Family(p.getKey(), p.getValue()));
families.forEach(f -> System.out.println(f));
}
Ответ 3
Это можно сделать с помощью collapse
с StreamEx
StreamEx.of(queryBuilder.createQuery(
"SELECT f.name, m.name"
+ " FROM Family f JOIN Member m on m.family_id = f.id"
+ " ORDER BY f.name"))
.collapse((a, b) -> a.string(0).equals(b.string(0)), Collectors.toList())
.map(l -> new Family(l.get(0).string(0), StreamEx.of(l).map(r -> r.string(1)).toList()))
.forEach(System.out::println);