Devam ettirilebilir iterator: state kaydet, birleştir, dosyadan kesintisiz oku

var it = new ListResumableIterator<>(List.of("a", "b"));
it.next()     -> "a"
it.next()     -> "b"
it.hasNext()  -> false
it.next()     -> NoSuchElementException

IteratorState getState();
void setState(IteratorState state);

it.next();                                   // "x"
String saved = IteratorState.encode(it.getState());   // "L 1"
var fresh = new ListResumableIterator<>(List.of("x", "y", "z"));
fresh.setState(IteratorState.decode(saved));
fresh.next()  -> "y"

var chain = new ChainedResumableIterator<>(List.of(
        new ListResumableIterator<>(List.of(1)),
        new ChainedResumableIterator<>(List.of(
                new ListResumableIterator<>(List.of(2)),
                new ListResumableIterator<>(List.of(3, 4)))),
        new ListResumableIterator<>(List.of(5, 6))));
chain.next(); chain.next(); chain.next();      // 1, 2, 3
IteratorState.encode(chain.getState())  -> "C 1 3 L 1 C 1 2 L 1 L 1 L 0"

var processor = new ResumableFileProcessor(checkpointFile, 2);   // her 2 satırda checkpoint
processor.run(input, line -> handle(line));   // 5. satırda çöktü (son checkpoint 4. satırdan sonra)
processor.run(input, line -> handle(line));   // 5, 6, 7'yi işler

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;

/**
 * Immutable snapshot of an iterator's position. Sealed, so encode/decode and setState
 * can switch over every possible shape and the compiler checks exhaustiveness.
 */
public sealed interface IteratorState {

    /** Level 2: position in a list = index of the next element. */
    record ListState(int nextIndex) implements IteratorState {
        public ListState {
            if (nextIndex < 0) throw new IllegalArgumentException("nextIndex < 0");
        }
    }

    /** Level 3: which child is current, plus every child's own state. */
    record ChainState(int current, List<IteratorState> children) implements IteratorState {
        public ChainState {
            children = List.copyOf(children);
        }
    }

    /** Level 4: byte offset of the next line in a file. */
    record FileState(long byteOffset) implements IteratorState {
        public FileState {
            if (byteOffset < 0) throw new IllegalArgumentException("byteOffset < 0");
        }
    }

    /** Level 2: text form for persisting, e.g. "C 1 2 L 3 F 120" (prefix notation, space separated). */
    static String encode(IteratorState state) {
        return switch (state) {
            case ListState(int nextIndex) -> "L " + nextIndex;
            case FileState(long byteOffset) -> "F " + byteOffset;
            case ChainState(int current, List<IteratorState> children) -> {
                var sb = new StringBuilder("C ").append(current).append(' ').append(children.size());
                children.forEach(child -> sb.append(' ').append(encode(child)));
                yield sb.toString();
            }
        };
    }

    static IteratorState decode(String text) {
        Iterator<String> tokens = Arrays.asList(text.trim().split("\\s+")).iterator();
        IteratorState state = decode(tokens);
        if (tokens.hasNext()) {
            throw new IllegalArgumentException("trailing tokens in state: " + text);
        }
        return state;
    }

    private static IteratorState decode(Iterator<String> tokens) {
        try {
            String tag = tokens.next();
            return switch (tag) {
                case "L" -> new ListState(Integer.parseInt(tokens.next()));
                case "F" -> new FileState(Long.parseLong(tokens.next()));
                case "C" -> {
                    int current = Integer.parseInt(tokens.next());
                    int size = Integer.parseInt(tokens.next());
                    List<IteratorState> children = new ArrayList<>(size);
                    for (int i = 0; i < size; i++) {
                        children.add(decode(tokens));
                    }
                    yield new ChainState(current, children);
                }
                default -> throw new IllegalArgumentException("unknown state tag: " + tag);
            };
        } catch (NoSuchElementException | NumberFormatException e) {
            throw new IllegalArgumentException("malformed state", e);
        }
    }
}

import java.util.Iterator;

/** An iterator whose position can be saved and restored. */
public interface ResumableIterator<T> extends Iterator<T> {

    /** Snapshot of the current position; later calls to next() do not change it. */
    IteratorState getState();

    /** Moves to a position previously returned by getState() of a compatible iterator. */
    void setState(IteratorState state);
}

import java.util.List;
import java.util.NoSuchElementException;

/** Levels 1-2: iterator over a list with save/restore. */
public final class ListResumableIterator<T> implements ResumableIterator<T> {

    private final List<T> items;
    private int nextIndex;

    public ListResumableIterator(List<T> items) {
        this.items = List.copyOf(items);   // a later change to the caller's list must not move us
    }

    @Override
    public boolean hasNext() {
        return nextIndex < items.size();
    }

    @Override
    public T next() {
        if (!hasNext()) {
            throw new NoSuchElementException();
        }
        return items.get(nextIndex++);
    }

    @Override
    public IteratorState getState() {
        return new IteratorState.ListState(nextIndex);
    }

    @Override
    public void setState(IteratorState state) {
        if (!(state instanceof IteratorState.ListState(int index))) {
            throw new IllegalArgumentException("not a list state: " + state);
        }
        if (index > items.size()) {
            throw new IllegalArgumentException("index " + index + " beyond size " + items.size());
        }
        nextIndex = index;
    }
}

import java.util.List;
import java.util.NoSuchElementException;

/**
 * Level 3: walks several resumable iterators one after another. Children can be chains
 * themselves, so nested structures compose without special cases.
 */
public final class ChainedResumableIterator<T> implements ResumableIterator<T> {

    private final List<ResumableIterator<T>> children;
    private int current;

    public ChainedResumableIterator(List<? extends ResumableIterator<T>> children) {
        this.children = List.copyOf(children);
    }

    @Override
    public boolean hasNext() {
        // Skip exhausted (or empty) children; afterwards current points at a child with data, or past the end.
        while (current < children.size() && !children.get(current).hasNext()) {
            current++;
        }
        return current < children.size();
    }

    @Override
    public T next() {
        if (!hasNext()) {
            throw new NoSuchElementException();
        }
        return children.get(current).next();
    }

    @Override
    public IteratorState getState() {
        return new IteratorState.ChainState(current, children.stream().map(ResumableIterator::getState).toList());
    }

    @Override
    public void setState(IteratorState state) {
        if (!(state instanceof IteratorState.ChainState(int index, List<IteratorState> childStates))) {
            throw new IllegalArgumentException("not a chain state: " + state);
        }
        if (childStates.size() != children.size() || index < 0 || index > children.size()) {
            throw new IllegalArgumentException("state does not match this chain");
        }
        // Children validate their own part; restore all of them so earlier and later children are consistent too.
        for (int i = 0; i < children.size(); i++) {
            children.get(i).setState(childStates.get(i));
        }
        current = index;
    }
}

import java.io.BufferedInputStream;
import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.NoSuchElementException;

/**
 * Level 4: reads UTF-8 lines from a file. The state is a byte offset (not a line number and
 * not a char count), so resuming is a seek and does not re-read the beginning of the file.
 */
public final class FileLineIterator implements ResumableIterator<String>, Closeable {

    private final FileChannel channel;
    private InputStream in;
    private long offset;                      // bytes consumed so far = start of the next line

    public FileLineIterator(Path path) throws IOException {
        this.channel = FileChannel.open(path, StandardOpenOption.READ);
        seek(0);
    }

    @Override
    public boolean hasNext() {
        try {
            return offset < channel.size();
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    @Override
    public String next() {
        if (!hasNext()) {
            throw new NoSuchElementException();
        }
        try {
            var line = new ByteArrayOutputStream();
            int b;
            while ((b = in.read()) != -1) {
                offset++;
                if (b == '\n') {
                    break;
                }
                line.write(b);
            }
            byte[] bytes = line.toByteArray();
            int length = bytes.length;
            if (length > 0 && bytes[length - 1] == '\r') {
                length--;                     // tolerate Windows line endings
            }
            // Split on bytes, decode after: '\n' never appears inside a multi-byte UTF-8 sequence.
            return new String(bytes, 0, length, StandardCharsets.UTF_8);
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    @Override
    public IteratorState getState() {
        return new IteratorState.FileState(offset);
    }

    @Override
    public void setState(IteratorState state) {
        if (!(state instanceof IteratorState.FileState(long target))) {
            throw new IllegalArgumentException("not a file state: " + state);
        }
        try {
            if (target > channel.size()) {
                throw new IllegalArgumentException("offset " + target + " beyond file size " + channel.size());
            }
            if (target > 0 && byteAt(target - 1) != '\n') {
                throw new IllegalArgumentException("offset " + target + " is not at a line boundary");
            }
            seek(target);
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    @Override
    public void close() throws IOException {
        channel.close();
    }

    private void seek(long target) throws IOException {
        channel.position(target);
        // BufferedInputStream reads ahead, so the channel position runs ahead of "offset"; that is why we count bytes ourselves.
        in = new BufferedInputStream(Channels.newInputStream(channel));
        offset = target;
    }

    private byte byteAt(long position) throws IOException {
        ByteBuffer one = ByteBuffer.allocate(1);
        channel.read(one, position);          // positional read, does not move the channel
        return one.get(0);
    }
}

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption;
import java.util.Optional;
import java.util.function.Consumer;

/**
 * Level 4: processes a file line by line and survives a crash by checkpointing the iterator state.
 * Delivery is at-least-once: lines after the last checkpoint are processed again after a restart.
 */
public final class ResumableFileProcessor {

    private final Path checkpointFile;
    private final int checkpointEvery;

    public ResumableFileProcessor(Path checkpointFile, int checkpointEvery) {
        if (checkpointEvery <= 0) {
            throw new IllegalArgumentException("checkpointEvery must be positive");
        }
        this.checkpointFile = checkpointFile;
        this.checkpointEvery = checkpointEvery;
    }

    /** Returns how many lines were handled in this run. */
    public int run(Path input, Consumer<String> handler) throws IOException {
        try (var lines = new FileLineIterator(input)) {
            loadCheckpoint().ifPresent(lines::setState);
            int handled = 0;
            while (lines.hasNext()) {
                handler.accept(lines.next());  // may throw: the checkpoint stays at the last saved position
                handled++;
                if (handled % checkpointEvery == 0) {
                    saveCheckpoint(lines.getState());
                }
            }
            saveCheckpoint(lines.getState());
            return handled;
        }
    }

    Optional<IteratorState> loadCheckpoint() throws IOException {
        if (!Files.exists(checkpointFile)) {
            return Optional.empty();
        }
        return Optional.of(IteratorState.decode(Files.readString(checkpointFile, StandardCharsets.UTF_8)));
    }

    /** Write to a temp file, then atomically rename: a crash never leaves a half-written checkpoint. */
    void saveCheckpoint(IteratorState state) throws IOException {
        Path temp = checkpointFile.resolveSibling(checkpointFile.getFileName() + ".tmp");
        Files.writeString(temp, IteratorState.encode(state), StandardCharsets.UTF_8,
                StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING,
                StandardOpenOption.WRITE, StandardOpenOption.DSYNC);
        Files.move(temp, checkpointFile, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
    }
}

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;

public final class ResumableIteratorTest {

    interface TestCase { void run() throws Exception; }

    public static void main(String[] args) throws Exception {
        Map<String, TestCase> tests = new LinkedHashMap<>();
        tests.put("level1_iteratesInOrderThenThrows", ResumableIteratorTest::level1_iteratesInOrderThenThrows);
        tests.put("level2_restoreRewindsAndReplays", ResumableIteratorTest::level2_restoreRewindsAndReplays);
        tests.put("level2_stateSurvivesEncodeDecodeIntoNewInstance", ResumableIteratorTest::level2_stateSurvivesEncodeDecodeIntoNewInstance);
        tests.put("level2_invalidStatesRejected", ResumableIteratorTest::level2_invalidStatesRejected);
        tests.put("level3_chainSkipsEmptyChildren", ResumableIteratorTest::level3_chainSkipsEmptyChildren);
        tests.put("level3_nestedChainResumesMidway", ResumableIteratorTest::level3_nestedChainResumesMidway);
        tests.put("level4_readsUtf8CrLfAndLastLineWithoutNewline", ResumableIteratorTest::level4_readsUtf8CrLfAndLastLineWithoutNewline);
        tests.put("level4_resumeFromByteOffset", ResumableIteratorTest::level4_resumeFromByteOffset);
        tests.put("level4_crashAndResumeIsAtLeastOnce", ResumableIteratorTest::level4_crashAndResumeIsAtLeastOnce);

        int passed = 0;
        for (var test : tests.entrySet()) {
            test.getValue().run();
            System.out.println("PASS " + test.getKey());
            passed++;
        }
        System.out.println(passed + "/" + tests.size() + " tests passed");
    }

    static void level1_iteratesInOrderThenThrows() {
        var it = new ListResumableIterator<>(List.of("a", "b"));
        check(drain(it).equals(List.of("a", "b")), "order");
        expectThrows(NoSuchElementException.class, it::next);
    }

    static void level2_restoreRewindsAndReplays() {
        var it = new ListResumableIterator<>(List.of(1, 2, 3, 4));
        it.next();
        IteratorState saved = it.getState();
        check(drain(it).equals(List.of(2, 3, 4)), "rest");
        it.setState(saved);
        check(drain(it).equals(List.of(2, 3, 4)), "replayed from saved state");
    }

    static void level2_stateSurvivesEncodeDecodeIntoNewInstance() {
        var source = new ArrayList<>(List.of("x", "y", "z"));
        var first = new ListResumableIterator<>(source);
        first.next();
        source.clear();                                        // caller's list changes; iterator has its own copy
        String persisted = IteratorState.encode(first.getState());
        check(persisted.equals("L 1"), "encoded: " + persisted);
        var second = new ListResumableIterator<>(List.of("x", "y", "z"));
        second.setState(IteratorState.decode(persisted));
        check(drain(second).equals(List.of("y", "z")), "resumed in a new instance");
    }

    static void level2_invalidStatesRejected() {
        var it = new ListResumableIterator<>(List.of("a"));
        expectThrows(IllegalArgumentException.class, () -> it.setState(new IteratorState.FileState(0)));
        expectThrows(IllegalArgumentException.class, () -> it.setState(new IteratorState.ListState(2)));
        expectThrows(IllegalArgumentException.class, () -> IteratorState.decode("C 0 2 L 1"));
        expectThrows(IllegalArgumentException.class, () -> IteratorState.decode("X 1"));
    }

    static void level3_chainSkipsEmptyChildren() {
        var chain = new ChainedResumableIterator<>(List.of(
                new ListResumableIterator<String>(List.of()),
                new ListResumableIterator<>(List.of("a")),
                new ListResumableIterator<String>(List.of()),
                new ListResumableIterator<>(List.of("b", "c"))));
        check(drain(chain).equals(List.of("a", "b", "c")), "empty children skipped");
        check(!chain.hasNext(), "exhausted");
    }

    static void level3_nestedChainResumesMidway() {
        var chain = nestedChain();
        check(chain.next() == 1 && chain.next() == 2 && chain.next() == 3, "first three");
        String persisted = IteratorState.encode(chain.getState());
        check(persisted.equals("C 1 3 L 1 C 1 2 L 1 L 1 L 0"), "encoded: " + persisted);
        check(drain(chain).equals(List.of(4, 5, 6)), "rest of first run");

        var restored = nestedChain();                          // e.g. after a restart
        restored.setState(IteratorState.decode(persisted));
        check(drain(restored).equals(List.of(4, 5, 6)), "resumed: " + persisted);
    }

    static ChainedResumableIterator<Integer> nestedChain() {
        var inner = new ChainedResumableIterator<>(List.of(
                new ListResumableIterator<>(List.of(2)),
                new ListResumableIterator<>(List.of(3, 4))));
        return new ChainedResumableIterator<>(List.of(
                new ListResumableIterator<>(List.of(1)),
                inner,
                new ListResumableIterator<>(List.of(5, 6))));
    }

    static void level4_readsUtf8CrLfAndLastLineWithoutNewline() throws IOException {
        Path file = writeTemp("şeker\r\nçay\n\nıhlamur");
        try (var it = new FileLineIterator(file)) {
            check(drain(it).equals(List.of("şeker", "çay", "", "ıhlamur")), "lines");
        }
    }

    static void level4_resumeFromByteOffset() throws IOException {
        Path file = writeTemp("ğ1\nğ2\nğ3\n");                  // 'ğ' is 2 bytes in UTF-8
        IteratorState afterFirst;
        try (var it = new FileLineIterator(file)) {
            it.next();
            afterFirst = it.getState();
            check(afterFirst.equals(new IteratorState.FileState(4)), "byte offset, not char offset: " + afterFirst);
        }
        try (var it = new FileLineIterator(file)) {
            it.setState(afterFirst);
            check(drain(it).equals(List.of("ğ2", "ğ3")), "resumed");
            expectThrows(IllegalArgumentException.class, () -> it.setState(new IteratorState.FileState(2)));
            expectThrows(IllegalArgumentException.class, () -> it.setState(new IteratorState.FileState(999)));
        }
    }

    static void level4_crashAndResumeIsAtLeastOnce() throws IOException {
        Path input = writeTemp("1\n2\n3\n4\n5\n6\n7\n");
        Path checkpoint = input.resolveSibling("checkpoint.txt");
        var processor = new ResumableFileProcessor(checkpoint, 2);
        List<String> seen = new ArrayList<>();

        try {
            processor.run(input, line -> {
                seen.add(line);
                if (line.equals("5")) throw new IllegalStateException("simulated crash");
            });
            throw new AssertionError("crash expected");
        } catch (IllegalStateException expected) {
            // checkpoint was saved after line 4
        }
        int secondRun = processor.run(input, seen::add);
        check(secondRun == 3, "second run handled 5, 6, 7");
        check(seen.equals(List.of("1", "2", "3", "4", "5", "5", "6", "7")), "line 5 twice: " + seen);
        check(processor.run(input, seen::add) == 0, "finished file is not processed again");
    }

    // ---------- helpers ----------

    static Path writeTemp(String content) throws IOException {
        Path dir = Files.createTempDirectory("resumable");
        Path file = dir.resolve("input.txt");
        Files.writeString(file, content, StandardCharsets.UTF_8);
        return file;
    }

    static <T> List<T> drain(ResumableIterator<T> it) {
        List<T> out = new ArrayList<>();
        while (it.hasNext()) out.add(it.next());
        return out;
    }

    static void check(boolean condition, String message) {
        if (!condition) {
            throw new AssertionError(message);
        }
    }

    static void expectThrows(Class<? extends Throwable> expected, Runnable action) {
        try {
            action.run();
        } catch (Throwable t) {
            if (expected.isInstance(t)) {
                return;
            }
            throw new AssertionError("expected " + expected.getSimpleName() + " but got " + t);
        }
        throw new AssertionError("expected " + expected.getSimpleName() + " but nothing was thrown");
    }
}