package stexfires.core.modifier;

import stexfires.core.Record;
import stexfires.core.message.RecordMessage;
import stexfires.core.record.KeyRecord;

import java.util.List;
import java.util.Objects;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author Mathias Kalb
 * @since 0.1
 */
public class GroupModifier<T extends Record, R extends Record> implements RecordStreamModifier<T, R> {

    protected final Function<? super T, ?> groupByClassifier;
    protected final Predicate<List<? super T>> havingPredicate;
    protected final Function<List<T>, ? extends R> aggregateFunction;

    public GroupModifier(Function<? super T, ?> groupByClassifier,
                         Function<List<T>, ? extends R> aggregateFunction) {
        this(groupByClassifier, list -> true, aggregateFunction);
    }

    public GroupModifier(Function<? super T, ?> groupByClassifier,
                         Predicate<List<? super T>> havingPredicate,
                         Function<List<T>, ? extends R> aggregateFunction) {
        Objects.requireNonNull(groupByClassifier);
        Objects.requireNonNull(havingPredicate);
        Objects.requireNonNull(aggregateFunction);
        this.groupByClassifier = groupByClassifier;
        this.havingPredicate = havingPredicate;
        this.aggregateFunction = aggregateFunction;
    }

    public static <T extends Record> Function<? super T, String> groupByCategory() {
        return Record::getCategory;
    }

    public static <T extends Record> Function<? super T, String> groupByMessage(RecordMessage<? super T> recordMessage) {
        return recordMessage.asFunction();
    }

    public static <T extends KeyRecord> Function<? super T, String> groupByKey() {
        return KeyRecord::getValueOfKeyField;
    }

    public static <T extends Record> Predicate<List<? super T>> sizeGreaterThan(int size) {
        return list -> list.size() > size;
    }

    @Override
    public Stream<R> modify(Stream<T> recordStream) {
        return recordStream
                .collect(Collectors.collectingAndThen(Collectors.groupingBy(groupByClassifier),
                        r -> r.values().stream()))
                .filter(havingPredicate)
                .map(aggregateFunction);
    }

}

@@ -141,16 +141,146 @@
 yRecord;
+%0Aimport stexfires.core.record.SingleRecord;%0Aimport stexfires.core.record.StandardRecord;%0Aimport stexfires.core.record.ValueRecord;
 %0A%0Aimport
@@ -358,16 +358,56 @@
 nction;%0A
+import java.util.function.IntPredicate;%0A
 import j
@@ -508,16 +508,45 @@
 m;%0A%0A/**%0A
+ * Aggregator / Group By.%0A *%0A
  * @auth
@@ -2089,112 +2089,1549 @@
 nds 
-Record%3E Predicate%3CList%3C? super T%3E%3E sizeGreaterThan(int size) %7B%0A        return list -%3E list.size() %3E size
+ValueRecord%3E Function%3C? super T, String%3E groupByValue() %7B%0A        return ValueRecord::getValueOfValueField;%0A    %7D%0A%0A    public static %3CT extends Record%3E Function%3C? super T, String%3E groupByValue(int index) %7B%0A        return r -%3E r.getValueAt(index);%0A    %7D%0A%0A    public static %3CT extends Record%3E Predicate%3CList%3C? super T%3E%3E havingSize(IntPredicate sizePredicate) %7B%0A        Objects.requireNonNull(sizePredicate);%0A        return list -%3E sizePredicate.test(list.size());%0A    %7D%0A%0A    public static %3CT extends Record%3E Predicate%3CList%3C? super T%3E%3E havingSizeGreaterThan(int size) %7B%0A        return list -%3E list.size() %3E size;%0A    %7D%0A%0A    public static %3CT extends Record%3E Function%3CList%3CT%3E, ValueRecord%3E aggregateToValue(Function%3CList%3CT%3E, String%3E categoryFunction,%0A                                                                                     Function%3CList%3CT%3E, String%3E valueFunction) %7B%0A        Objects.requireNonNull(categoryFunction);%0A        Objects.requireNonNull(valueFunction);%0A        return list -%3E new SingleRecord(categoryFunction.apply(list), null, valueFunction.apply(list));%0A    %7D%0A%0A    public static %3CT extends Record%3E Function%3CList%3CT%3E, Record%3E aggregateToValues(Function%3CList%3CT%3E, String%3E categoryFunction,%0A                                                                                 Function%3CList%3CT%3E, List%3CString%3E%3E valuesFunction) %7B%0A        Objects.requireNonNull(categoryFunction);%0A        Objects.requireNonNull(valuesFunction);%0A        return list -%3E new StandardRecord(categoryFunction.apply(list), null, valuesFunction.apply(list))
 ;%0A  

Add more static methods