// Path: src/main/java/net/kimleo/rec/v2/model/Source.java
// 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);
//     }
// }
// 
// Path: src/main/java/net/kimleo/rec/v2/model/Target.java
// 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);
//     }
// 
// }
// 
// Path: src/main/java/net/kimleo/rec/v2/model/Tee.java
// 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);
//     }
// }


import net.kimleo.rec.v2.model.Source;
import net.kimleo.rec.v2.model.Target;
import net.kimleo.rec.v2.model.Tee;
import java.util.function.Predicate;
import java.util.stream.Stream;
    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
// Path: src/main/java/net/kimleo/rec/v2/model/Source.java
// 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);
//     }
// }
// 
// Path: src/main/java/net/kimleo/rec/v2/model/Target.java
// 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);
//     }
// 
// }
// 
// Path: src/main/java/net/kimleo/rec/v2/model/Tee.java
// 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);
//     }
// }


// Path: src/main/java/net/kimleo/rec/v2/model/impl/ReactiveTee.java
import net.kimleo.rec.v2.model.Source;
import net.kimleo.rec.v2.model.Target;
import net.kimleo.rec.v2.model.Tee;
import java.util.function.Predicate;
import java.util.stream.Stream;

    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) {