public interface Mapped<T> {
    T get(String field);

    List<String> keys();
} public class ParseConfig {
    public final char delimiter;
    public final char escape;

    public static final ParseConfig DEFAULT = new ParseConfig();
    public static final ParseConfig BY_PIPE = new ParseConfig('|');

    public ParseConfig(char delimiter, char escape) {
        this.delimiter = delimiter;
        this.escape = escape;
    }

    public ParseConfig(char delimiter) {
        this(delimiter, '"');
    }

    public ParseConfig() {
        this(',', '"');
    }
} public interface ExecutionContext {
    void commit();
    void persist(Throwable causedBy);

    ExecutionContext restart();

    boolean isNative();
    boolean isCloud();

    Context jsContext();
} public interface Source<T> {
    Stream<T> stream();
    default Source<T> tee(Tee<T> tee) {
        return () -> this.stream().map(tee::emit);
    }

    default Source<T> filter(Predicate<T> predicate) {
        return () -> this.stream().filter(predicate);
    }

    default Source<T> skip(int n) {
        return from(stream().skip(n));
    }

    static <T> Source<T> from(Stream<T> records) {
        return () -> records;
    }

    default void to(Target<T> target) {
        target.putAll(this);
    }
} public interface Target<T> {
    void put(T record);

    default Target<T> tee(Tee<T> tee) {
        return record -> this.put(tee.emit(record));
    }

    default void putAll(Source<T> source) {
        source.stream().forEach(this::put);
    }

} public interface Tee<T> {
    T emit(T record);

    default Source<T> source() {
        return Source.from(Stream.empty());
    }
    default void to(Target<T> target) {
        source().to(target);
    }
} public class CSVSource implements Source<Mapped<String>> {

    private final BufferedReader reader;
    private final Accessor<String> accessor;
    private final SimpleParser csvParser;
    private final int skipLimit;

    private CSVSource(BufferedReader reader,
                      Accessor<String> accessor,
                      SimpleParser csvParser,
                      int skipLimit) {
        this.reader = reader;
        this.accessor = accessor;
        this.csvParser = csvParser;
        this.skipLimit = skipLimit;
    }

    public CSVSource(Reader reader, String accessors, ParseConfig config) {
        accessor = new Accessor<>(new SimpleParser().parse(accessors).getValues().toArray(new String[]{}));
        csvParser = new SimpleParser(config);
        this.reader = new BufferedReader(reader);
        skipLimit = 0;
    }

    @Override
    public Stream<Mapped<String>> stream() {
        return reader.lines().skip(skipLimit).map(line -> {
            return accessor.of(csvParser.parse(line));
        });
    }

    @Override
    public Source<Mapped<String>> skip(int n) {
        return new CSVSource(reader, accessor, csvParser, n);
    }
} public class CollectTee<T> implements Tee<T> {
    private final Collection<T> collection;


    public CollectTee(Collection<T> collection) {
        this.collection = collection;
    }

    public Collection<T> collect() {
        return collection;
    }

    @Override
    public T emit(T record) {
        collection.add(record);
        return record;
    }

    @Override
    public Source<T> source() {
        return Source.from(collection.stream());
    }
} public class ItemCounterTee<T> implements Tee<T> {

    private final Predicate<T> predicate;
    private int count;

    public ItemCounterTee(Predicate<T> predicate) {
        this.predicate = predicate;
        this.count = 0;
    }

    @Override
    public T emit(T record) {
        if(predicate.test(record)) {
            count ++;
        }
        return record;
    }

    public int getCount() {
        return count;
    }
} public class ReactiveTee<T> implements Tee<T>, Source<T> {
    private Target<T> target = null;
    protected final Tee<T> parent;

    public ReactiveTee(Tee<T> parent) {
        this.parent = parent;
    }

    @Override
    public T emit(T record) {
        return toTarget(record);
    }

    @Override
    public void to(Target<T> target) {
        this.target = target;
    }

    @Override
    public Source<T> filter(Predicate<T> predicate) {
        return new FilterNode(this, predicate);
    }

    @Override
    public Source<T> tee(Tee<T> tee) {
        return new TeeNode(this, tee);
    }

    @Override
    public Source<T> skip(int n) {
        return new SkipNode(this, n);
    }

    @Override
    public Stream<T> stream() {
        throw new IllegalStateException("Reactive tee has no source");
    }

    protected T toTarget(T item) {
        if (target != null) {
            target.put(item);
        }
        return item;
    }

    class FilterNode extends ReactiveTee<T>{
        private final Predicate<T> filter;

        FilterNode(Tee<T> parent, Predicate<T> filter) {
            super(parent);
            this.filter = filter;
        }

        @Override
        public T emit(T record) {
            T result = this.parent.emit(record);
            if (result != null) {
                if (filter.test(result)) {
                    return this.toTarget(result);
                }
            }
            return null;
        }
    }

    class TeeNode extends ReactiveTee<T> {

        private final Tee<T> tee;

        TeeNode(Tee<T> parent, Tee<T> tee) {
            super(parent);
            this.tee = tee;
        }

        @Override
        public T emit(T record) {
            T result = this.parent.emit(record);
            if (result != null) {
                return this.toTarget(tee.emit(result));
            }
            return null;
        }
    }

    class SkipNode extends ReactiveTee<T> {
        private int skips;

        SkipNode(Tee<T> reactive, int n) {
            super(reactive);
            this.skips = n;
        }

        @Override
        public T emit(T record) {

            if (skips > 0) {
                this.skips --;
                this.parent.emit(record);
                return null;
            } else {
                return this.toTarget(this.parent.emit(record));
            }
        }
    }
} public class ResultSetSource implements Source<Mapped<String>> {
    private final ResultSet rs;
    private static final Logger LOGGER = LoggerFactory.getLogger(ResultSetSource.class);

    public ResultSetSource(ResultSet rs) {
        this.rs = rs;
    }

    @Override
    public Stream<Mapped<String>> stream() {
        return StreamSupport.stream(new GeneratingSpliteratorAdapter<Mapped<String>>(() -> {
            try {
                if (rs.next()) {
                    LOGGER.info(format("Get next item in result set #[%d]", rs.hashCode()));
                    return new ResultSetMapper(rs);
                } else {
                    return null;
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
            return null;
        }),false);
    }
} public class JSRecord implements Scriptable, Mapped<String> {

    private final Mapped<String> record;
    Scriptable parent = this;

    public JSRecord(Mapped<String> record) {
        this.record = record;
    }

    @Override
    public String getClassName() {
        return "JSRecord";
    }

    @Override
    public Object get(String name, Scriptable start) {
        return has(name, start) ? record.get(name) : null;
    }

    @Override
    public Object get(int index, Scriptable start) {
        return null;
    }

    @Override
    public boolean has(String name, Scriptable start) {
        return record.keys().contains(name);
    }

    @Override
    public boolean has(int index, Scriptable start) {
        return false;
    }

    @Override
    public void put(String name, Scriptable start, Object value) {
        throw new UnsupportedOperationException();
    }

    @Override
    public void put(int index, Scriptable start, Object value) {
        throw new UnsupportedOperationException();
    }

    @Override
    public void delete(String name) {
        throw new UnsupportedOperationException();
    }

    @Override
    public void delete(int index) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Scriptable getPrototype() {
        return parent;
    }

    @Override
    public void setPrototype(Scriptable prototype) {
        parent = prototype;
    }

    @Override
    public Scriptable getParentScope() {
        return parent;
    }

    @Override
    public void setParentScope(Scriptable parent) {
        this.parent = parent;
    }

    @Override
    public Object[] getIds() {
        return record.keys().toArray();
    }

    @Override
    public Object getDefaultValue(Class<?> hint) {
        return null;
    }

    @Override
    public boolean hasInstance(Scriptable instance) {
        return false;
    }

    @Override
    public String get(String field) {
        return record.get(field);
    }

    @Override
    public List<String> keys() {
        return record.keys();
    }
} public class State {
    public Object state;

    public State(Object state) {
        this.state = state;
    }

    public Object get() { return state; }
    public State set(Object state) { this.state = state; return this;}

    public Object getValue() { return state; }

}
import net.kimleo.rec.common.concept.Mapped;
import net.kimleo.rec.sepval.parser.ParseConfig;
import net.kimleo.rec.v2.execution.ExecutionContext;
import net.kimleo.rec.v2.model.Source;
import net.kimleo.rec.v2.model.Target;
import net.kimleo.rec.v2.model.Tee;
import net.kimleo.rec.v2.model.impl.CSVSource;
import net.kimleo.rec.v2.model.impl.CollectTee;
import net.kimleo.rec.v2.model.impl.ItemCounterTee;
import net.kimleo.rec.v2.model.impl.ReactiveTee;
import net.kimleo.rec.v2.model.impl.ResultSetSource;
import net.kimleo.rec.v2.scripting.model.JSRecord;
import net.kimleo.rec.v2.scripting.model.State;
import org.mozilla.javascript.Context;
import org.mozilla.javascript.Function;
import org.mozilla.javascript.Scriptable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Reader;
import java.sql.ResultSet;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import static java.lang.String.format;
package net.kimleo.rec.v2.scripting.module;




public class Rec {
    private static final Logger LOGGER = LoggerFactory.getLogger(Rec.class);
    protected Context jsContext;

    public Rec(ExecutionContext context) {
        this.jsContext = context.jsContext();
    }

    // Wrappers
    public <T> Consumer<T> action(Function function) {
        LOGGER.info(format("Action wrapper created for Function #[%d]", function.hashCode()));
        return (object) ->
                function.call(jsContext, function.getParentScope(), null, new Object[]{ wrap(object) });
    }

    public  <T> Predicate<T> pred(Function function) {
        LOGGER.info(format("Predicate wrapper created for Function #[%d]", function.hashCode()));
        return (func) ->
                (boolean) function.call(jsContext, function.getParentScope(), null, new Object[]{ wrap(func) });
    }

    @SuppressWarnings("unchecked")
    private static <T> Object wrap(T obj) {
        if (obj instanceof Mapped) {
            return new JSRecord((Mapped<String>) obj);