// Path: src/main/java/net/kimleo/rec/common/Pair.java
// public class Pair<K, V> {
//     public final K first;
//     public final V second;
// 
// 
//     public Pair(K first, V second) {
//         this.first = first;
//         this.second = second;
//     }
// 
//     public K getFirst() {
//         return first;
//     }
// 
//     public V getSecond() {
//         return second;
//     }
// 
// }
// 
// Path: src/main/java/net/kimleo/rec/common/concept/Mapped.java
// public interface Mapped<T> {
//     T get(String field);
// 
//     List<String> keys();
// }
// 
// Path: src/main/java/net/kimleo/rec/common/exception/ResourceAccessException.java
// public class ResourceAccessException extends RuntimeException {
//     public ResourceAccessException(String message, Throwable ex) {
//         super(message, ex);
//     }
// }
// 
// 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/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/stream/adapter/GeneratingSpliteratorAdapter.java
// public class GeneratingSpliteratorAdapter<T> extends Spliterators.AbstractSpliterator<T> {
// 
// 
//     private final Supplier<T> supplier;
// 
//     public GeneratingSpliteratorAdapter(Supplier<T> supplier) {
//         super(Long.MAX_VALUE, 0);
//         this.supplier = supplier;
//     }
// 
//     @Override
//     public boolean tryAdvance(Consumer<? super T> action) {
//         T item = supplier.get();
//         if (item != null) {
//             action.accept(item);
//             return true;
//         }
// 
//         return false;
//     }
// }
// 
// Path: src/main/java/net/kimleo/rec/v2/utils/Records.java
// public static Pair<List<String>, Integer> decode(ByteBuffer buffer, int readPos, int expectedSize) {
// 
//     buffer.position(readPos);
//     int size = buffer.getInt();
//     if (size != expectedSize) return null;
//     ArrayList<String> items = new ArrayList<>(size);
// 
//     for (int i = 0; i < size; i++) {
//         int length = buffer.getInt();
//         byte[] bytes = new byte[length];
//         buffer.get(bytes, 0, length);
//         items.add(new String(bytes));
// 
//     }
// 
//     return new Pair<>(items, buffer.position() - readPos);
// }
// 
// Path: src/main/java/net/kimleo/rec/v2/utils/Records.java
// public static ByteBuffer encode(Mapped<String> record, List<String> keys) {
//     int size = keys.size();
//     ByteBuffer byteBuffer = ByteBuffer.allocate(sizeof(record, keys));
// 
//     byteBuffer.putInt(size);
//     for (String key : keys) {
//         String field = record.get(key);
//         byteBuffer.putInt(field.getBytes().length);
//         byteBuffer.put(field.getBytes());
//     }
//     return byteBuffer;
// }


import net.kimleo.rec.common.Pair;
import net.kimleo.rec.common.concept.Mapped;
import net.kimleo.rec.common.exception.ResourceAccessException;
import net.kimleo.rec.v2.model.Source;
import net.kimleo.rec.v2.model.Tee;
import net.kimleo.rec.v2.stream.adapter.GeneratingSpliteratorAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.List;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import static net.kimleo.rec.v2.utils.Records.decode;
import static net.kimleo.rec.v2.utils.Records.encode;

    private void persist() {
        try {
            int original = buffer.limit();
            buffer.limit(writePos).position(0);
            byte[] bytes = new byte[writePos];
            buffer.get(bytes);
            Files.newByteChannel(tempFile,
                    StandardOpenOption.WRITE,
                    StandardOpenOption.TRUNCATE_EXISTING).write(ByteBuffer.wrap(bytes));
            buffer.limit(original);
        } catch (IOException e) {
            throw new ResourceAccessException(String.format("Cannot access temp file: [%s]",
                    tempFile.getFileName()), e);
        }
    }

    static class BufferedCachingSource implements Source<Mapped<String>> {

        private final BufferedCachingTee tee;
        private final ByteBuffer buffer;
        private int readPos = 0;

        BufferedCachingSource(BufferedCachingTee tee) {
            this.tee = tee;
            this.buffer = tee.buffer;
        }

        @Override
        public Stream<Mapped<String>> stream() {
// Path: src/main/java/net/kimleo/rec/common/Pair.java
// public class Pair<K, V> {
//     public final K first;
//     public final V second;
// 
// 
//     public Pair(K first, V second) {
//         this.first = first;
//         this.second = second;
//     }
// 
//     public K getFirst() {
//         return first;
//     }
// 
//     public V getSecond() {
//         return second;
//     }
// 
// }
// 
// Path: src/main/java/net/kimleo/rec/common/concept/Mapped.java
// public interface Mapped<T> {
//     T get(String field);
// 
//     List<String> keys();
// }
// 
// Path: src/main/java/net/kimleo/rec/common/exception/ResourceAccessException.java
// public class ResourceAccessException extends RuntimeException {
//     public ResourceAccessException(String message, Throwable ex) {
//         super(message, ex);
//     }
// }
// 
// 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/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/stream/adapter/GeneratingSpliteratorAdapter.java
// public class GeneratingSpliteratorAdapter<T> extends Spliterators.AbstractSpliterator<T> {
// 
// 
//     private final Supplier<T> supplier;
// 
//     public GeneratingSpliteratorAdapter(Supplier<T> supplier) {
//         super(Long.MAX_VALUE, 0);
//         this.supplier = supplier;
//     }
// 
//     @Override
//     public boolean tryAdvance(Consumer<? super T> action) {
//         T item = supplier.get();
//         if (item != null) {
//             action.accept(item);
//             return true;
//         }
// 
//         return false;
//     }
// }
// 
// Path: src/main/java/net/kimleo/rec/v2/utils/Records.java
// public static Pair<List<String>, Integer> decode(ByteBuffer buffer, int readPos, int expectedSize) {
// 
//     buffer.position(readPos);
//     int size = buffer.getInt();
//     if (size != expectedSize) return null;
//     ArrayList<String> items = new ArrayList<>(size);
// 
//     for (int i = 0; i < size; i++) {
//         int length = buffer.getInt();
//         byte[] bytes = new byte[length];
//         buffer.get(bytes, 0, length);
//         items.add(new String(bytes));
// 
//     }
// 
//     return new Pair<>(items, buffer.position() - readPos);
// }
// 
// Path: src/main/java/net/kimleo/rec/v2/utils/Records.java
// public static ByteBuffer encode(Mapped<String> record, List<String> keys) {
//     int size = keys.size();
//     ByteBuffer byteBuffer = ByteBuffer.allocate(sizeof(record, keys));
// 
//     byteBuffer.putInt(size);
//     for (String key : keys) {
//         String field = record.get(key);
//         byteBuffer.putInt(field.getBytes().length);
//         byteBuffer.put(field.getBytes());
//     }
//     return byteBuffer;
// }


// Path: src/main/java/net/kimleo/rec/v2/model/impl/BufferedCachingTee.java
import net.kimleo.rec.common.Pair;
import net.kimleo.rec.common.concept.Mapped;
import net.kimleo.rec.common.exception.ResourceAccessException;
import net.kimleo.rec.v2.model.Source;
import net.kimleo.rec.v2.model.Tee;
import net.kimleo.rec.v2.stream.adapter.GeneratingSpliteratorAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.List;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import static net.kimleo.rec.v2.utils.Records.decode;
import static net.kimleo.rec.v2.utils.Records.encode;


    private void persist() {
        try {
            int original = buffer.limit();
            buffer.limit(writePos).position(0);
            byte[] bytes = new byte[writePos];
            buffer.get(bytes);
            Files.newByteChannel(tempFile,
                    StandardOpenOption.WRITE,
                    StandardOpenOption.TRUNCATE_EXISTING).write(ByteBuffer.wrap(bytes));
            buffer.limit(original);
        } catch (IOException e) {
            throw new ResourceAccessException(String.format("Cannot access temp file: [%s]",
                    tempFile.getFileName()), e);
        }
    }

    static class BufferedCachingSource implements Source<Mapped<String>> {

        private final BufferedCachingTee tee;
        private final ByteBuffer buffer;
        private int readPos = 0;

        BufferedCachingSource(BufferedCachingTee tee) {
            this.tee = tee;
            this.buffer = tee.buffer;
        }

        @Override
        public Stream<Mapped<String>> stream() {
            return StreamSupport.stream(new GeneratingSpliteratorAdapter<Mapped<String>>(() -> {