// Path: src/main/java/net/kimleo/rec/v2/execution/ExecutionContext.java
// public interface ExecutionContext {
//     void commit();
//     void persist(Throwable causedBy);
// 
//     ExecutionContext restart();
// 
//     boolean isNative();
//     boolean isCloud();
// 
//     Context jsContext();
// }
// 
// Path: src/main/java/net/kimleo/rec/v2/execution/RestartableSource.java
// public interface RestartableSource<T> extends Source<T> {
// 
//     ExecutionContext context();
// 
//     @Override
//     default void to(Target<T> target) {
//         try {
//             stream().forEach(target::put);
//         } catch (Throwable ex) {
//             context().persist(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/execution/impl/NativeExecutionContext.java
// public static NativeExecutionContext initialContext() {
//     return new NativeExecutionContext(0);
// }


import net.kimleo.rec.v2.execution.ExecutionContext;
import net.kimleo.rec.v2.execution.RestartableSource;
import net.kimleo.rec.v2.model.Source;
import java.util.stream.Stream;
import static net.kimleo.rec.v2.execution.impl.NativeExecutionContext.initialContext;
package net.kimleo.rec.v2.execution.impl;




public class CountBasedRestartableSource<T> implements RestartableSource<T> {

    private final Source<T> source;
    private final NativeExecutionContext context;

    public CountBasedRestartableSource(Source<T> source, NativeExecutionContext context) {
        this.source = source;
        this.context = context;
    }

    @Override
    public ExecutionContext context() {
        return this.context;
    }

    @Override
    public Source<T> skip(int n) {
        return source.skip(n);
    }

    @Override
    public Stream<T> stream() {
        return source.skip(context.state()).stream().map(it -> {
            context.commit();
            return it;
        });
    }

    public static <T> Source<T> from(Stream<T> stream) {
// Path: src/main/java/net/kimleo/rec/v2/execution/ExecutionContext.java
// public interface ExecutionContext {
//     void commit();
//     void persist(Throwable causedBy);
// 
//     ExecutionContext restart();
// 
//     boolean isNative();
//     boolean isCloud();
// 
//     Context jsContext();
// }
// 
// Path: src/main/java/net/kimleo/rec/v2/execution/RestartableSource.java
// public interface RestartableSource<T> extends Source<T> {
// 
//     ExecutionContext context();
// 
//     @Override
//     default void to(Target<T> target) {
//         try {
//             stream().forEach(target::put);
//         } catch (Throwable ex) {
//             context().persist(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/execution/impl/NativeExecutionContext.java
// public static NativeExecutionContext initialContext() {
//     return new NativeExecutionContext(0);
// }


// Path: src/main/java/net/kimleo/rec/v2/execution/impl/CountBasedRestartableSource.java
import net.kimleo.rec.v2.execution.ExecutionContext;
import net.kimleo.rec.v2.execution.RestartableSource;
import net.kimleo.rec.v2.model.Source;
import java.util.stream.Stream;
import static net.kimleo.rec.v2.execution.impl.NativeExecutionContext.initialContext;

package net.kimleo.rec.v2.execution.impl;




public class CountBasedRestartableSource<T> implements RestartableSource<T> {

    private final Source<T> source;
    private final NativeExecutionContext context;

    public CountBasedRestartableSource(Source<T> source, NativeExecutionContext context) {
        this.source = source;
        this.context = context;
    }

    @Override
    public ExecutionContext context() {
        return this.context;
    }

    @Override
    public Source<T> skip(int n) {
        return source.skip(n);
    }

    @Override
    public Stream<T> stream() {
        return source.skip(context.state()).stream().map(it -> {
            context.commit();
            return it;
        });
    }

    public static <T> Source<T> from(Stream<T> stream) {
        return new CountBasedRestartableSource<>(() -> stream, initialContext());