package org.textfiledatatools.core;

import org.textfiledatatools.core.producer.RecordProducer;
import org.textfiledatatools.core.producer.UncheckedProducerException;

import java.util.Comparator;
import java.util.Objects;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Stream;

/**
 * This class consists of {@code static} utility methods for operating on record streams.
 *
 * @author Mathias Kalb
 * @since 0.1
 */
public final class RecordStreams {

    private RecordStreams() {
    }

    public static Stream<Record> empty() {
        return Stream.empty();
    }

    public static <T extends Record> Stream<T> of(T record) {
        return Stream.of(record);
    }

    @SafeVarargs
    public static <T extends Record> Stream<T> of(T... records) {
        return Stream.of(records);
    }

    public static <T extends Record> Stream<T> produceStream(RecordProducer<T> recordProducer) throws UncheckedProducerException {
        return recordProducer.produceStream();
    }

    public static <T extends Record> Stream<T> generate(Supplier<T> recordSupplier) {
        return Stream.generate(recordSupplier);
    }

    public static <T extends Record> Stream<T> concat(Stream<T> firstRecordStream, Stream<T> secondRecordStream) {
        return Stream.concat(firstRecordStream, secondRecordStream);
    }

    @SafeVarargs
    public static <T extends Record> Stream<T> concat(Stream<T>... recordStreams) {
        // TODO Validate: concatWithFlatMap or concatWithReduce
        return concatWithFlatMap(recordStreams);
    }

    @SafeVarargs
    static <T extends Record> Stream<T> concatWithFlatMap(Stream<T>... recordStreams) {
        return Stream.of(recordStreams).flatMap(Function.identity());
    }

    @SafeVarargs
    static <T extends Record> Stream<T> concatWithReduce(Stream<T>... recordStreams) {
        return Stream.of(recordStreams).reduce(Stream.empty(), Stream::concat);
    }

    public static <T extends Record> Stream<T> distinct(Stream<T> stream,
                                                        Function<? super T, String> recordEqualsString) {
        Objects.requireNonNull(stream, "Parameter 'stream' must not be null");
        Objects.requireNonNull(recordEqualsString, "Parameter 'recordEqualsString' must not be null");
        //noinspection RedundantTypeArguments
        return stream.map(record -> new DistinctRecordWrapper<>(record, recordEqualsString.apply(record)))
                .distinct()
                .map(DistinctRecordWrapper<T>::getRecord);
    }

    public static <T extends Record> Stream<T> sorted(Stream<T> stream, Comparator<? super T> recordComparator) {
        Objects.requireNonNull(stream, "Parameter 'stream' must not be null");
        Objects.requireNonNull(recordComparator, "Parameter 'recordComparator' must not be null");
        return stream.sorted(recordComparator);
    }

    private static final class DistinctRecordWrapper<T extends Record> {

        private final T record;
        private final String equalsString;

        public DistinctRecordWrapper(T record, String equalsString) {
            this.record = record;
            this.equalsString = equalsString;
        }

        public T getRecord() {
            return record;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            DistinctRecordWrapper<?> that = (DistinctRecordWrapper<?>) o;
            return Objects.equals(equalsString, that.equalsString);
        }

        @Override
        public int hashCode() {
            return Objects.hash(equalsString);
        }

        @Override
        public String toString() {
            return "DistinctRecordWrapper{" +
                    "record=" + record +
                    ", equalsString='" + equalsString + '\'' +
                    '}';
        }
    }

}

@@ -30,16 +30,263 @@
 .core;%0A%0A
+import org.textfiledatatools.core.consumer.LoggerConsumer;%0Aimport org.textfiledatatools.core.consumer.RecordConsumer;%0Aimport org.textfiledatatools.core.consumer.UncheckedConsumerException;%0Aimport org.textfiledatatools.core.logger.SystemOutLogger;%0A
 import o
@@ -2211,32 +2211,50 @@
  public static %3C
+R extends Record, 
 T extends Record
@@ -2250,47 +2250,526 @@
 ends
+ R%3E
  Record
-%3E Stream%3CT%3E distinct(Stream%3CT%3E s
+Consumer%3CR%3E consume(Stream%3CT%3E recordStream,%0A                                                                            RecordConsumer%3CR%3E recordConsumer) throws UncheckedConsumerException %7B%0A        recordStream.forEachOrdered(recordConsumer::consume);%0A        return recordConsumer;%0A    %7D%0A%0A    public static void println(Stream%3CRecord%3E recordStream) %7B%0A        consume(recordStream, new LoggerConsumer%3C%3E(new SystemOutLogger()));%0A    %7D%0A%0A    public static %3CT extends Record%3E Stream%3CT%3E distinct(Stream%3CT%3E recordS
 trea
@@ -2900,33 +2900,39 @@
 .requireNonNull(
-s
+recordS
 tream, %22Paramete
@@ -2926,33 +2926,39 @@
 am, %22Parameter '
-s
+recordS
 tream' must not 
@@ -3128,25 +3128,31 @@
      return 
-s
+recordS
 tream.map(re
@@ -3388,17 +3388,23 @@
 ream%3CT%3E 
-s
+recordS
 tream, C
@@ -3475,17 +3475,23 @@
 NonNull(
-s
+recordS
 tream, %22
@@ -3501,17 +3501,23 @@
 ameter '
-s
+recordS
 tream' m
@@ -3649,17 +3649,23 @@
  return 
-s
+recordS
 tream.so

Add println() and consume()