<commit_before>package com.timgroup.eventstore.clock;

import com.timgroup.eventstore.api.EventReader;
import com.timgroup.eventstore.api.Position;
import com.timgroup.eventstore.api.PositionCodec;
import com.timgroup.eventstore.api.ResolvedEvent;
import com.timgroup.eventstore.api.StreamId;
import com.timgroup.eventstore.merging.NamedReaderWithCodec;

import javax.annotation.CheckReturnValue;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.stream.LongStream;
import java.util.stream.Stream;

import static com.timgroup.eventstore.api.EventRecord.eventRecord;
import static java.util.Objects.requireNonNull;

public class TimeReader implements EventReader, PositionCodec {
    private final Instant start;
    private final Duration accuracy;
    private final Clock clock;

    public TimeReader(Instant start, Duration accuracy, Clock clock) {
        this.start = requireNonNull(start);
        this.accuracy = requireNonNull(accuracy);
        this.clock = requireNonNull(clock);
    }

    @Nonnull
    public static NamedReaderWithCodec timePassedEventStream(Instant start, Duration accuracy, Clock clock) {
        TimeReader timeReader = new TimeReader(start, accuracy, clock);
        return new NamedReaderWithCodec("Clock", timeReader, timeReader);
    }

    @CheckReturnValue
    @Nonnull
    @Override
    public Stream<ResolvedEvent> readAllForwards(Position positionExclusive) {
        long now = clock.millis();
        long lastTime = ((TimePosition) positionExclusive).value;

        long numberOfEvents = (now - lastTime) / accuracy.toMillis();

        return LongStream.range(0, numberOfEvents).map(operand -> lastTime + ((operand + 1) * accuracy.toMillis())).mapToObj(time -> new ResolvedEvent(
                new TimePosition(time),
                eventRecord(
                        Instant.ofEpochMilli(time),
                        StreamId.streamId("time", ""),
                        -1,
                        "TimePassed",
                        new byte[0],
                        new byte[0]
                )));
    }

    @Nonnull
    @Override
    public Position emptyStorePosition() {
        return new TimePosition(start.toEpochMilli());
    }

    @Override
    public Position deserializePosition(String string) {
        return new TimePosition(Long.parseLong(string));
    }

    @Override
    public String serializePosition(Position position) {
        return Long.toString(((TimePosition) position).value);
    }

    @Override
    public String toString() {
        return "TimeReader{" +
                "start=" + start +
                ", accuracy=" + accuracy +
                ", clock=" + clock +
                '}';
    }

    private static final class TimePosition implements Position, Comparable<TimePosition> {
        private final long value;

        private TimePosition(long value) {<commit_msg>SteveH: use standard PositionCodec implementation with TimeReader<commit_after>package com.timgroup.eventstore.clock;

import com.timgroup.eventstore.api.EventReader;
import com.timgroup.eventstore.api.Position;
import com.timgroup.eventstore.api.PositionCodec;
import com.timgroup.eventstore.api.ResolvedEvent;
import com.timgroup.eventstore.api.StreamId;
import com.timgroup.eventstore.merging.NamedReaderWithCodec;

import javax.annotation.CheckReturnValue;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.stream.LongStream;
import java.util.stream.Stream;

import static com.timgroup.eventstore.api.EventRecord.eventRecord;
import static java.util.Objects.requireNonNull;

public final class TimeReader implements EventReader {
    private final Instant start;
    private final Duration accuracy;
    private final Clock clock;

    public TimeReader(Instant start, Duration accuracy, Clock clock) {
        this.start = requireNonNull(start);
        this.accuracy = requireNonNull(accuracy);
        this.clock = requireNonNull(clock);
    }

    @Nonnull
    public static NamedReaderWithCodec timePassedEventStream(Instant start, Duration accuracy, Clock clock) {
        TimeReader timeReader = new TimeReader(start, accuracy, clock);
        return new NamedReaderWithCodec("Clock", timeReader, TimePosition.CODEC);
    }

    @CheckReturnValue
    @Nonnull
    @Override
    public Stream<ResolvedEvent> readAllForwards(Position positionExclusive) {
        long now = clock.millis();
        long lastTime = ((TimePosition) positionExclusive).value;

        long numberOfEvents = (now - lastTime) / accuracy.toMillis();

        return LongStream.range(0, numberOfEvents).map(operand -> lastTime + ((operand + 1) * accuracy.toMillis())).mapToObj(time -> new ResolvedEvent(
                new TimePosition(time),
                eventRecord(
                        Instant.ofEpochMilli(time),
                        StreamId.streamId("time", ""),
                        -1,
                        "TimePassed",
                        new byte[0],
                        new byte[0]
                )));
    }

    @Nonnull
    @Override
    public Position emptyStorePosition() {
        return new TimePosition(start.toEpochMilli());
    }

    @Override
    public String toString() {
        return "TimeReader{" +
                "start=" + start +
                ", accuracy=" + accuracy +
                ", clock=" + clock +
                '}';
    }

    private static final class TimePosition implements Position, Comparable<TimePosition> {
        static final PositionCodec CODEC = PositionCodec.ofComparable(TimePosition.class,
                string -> new TimePosition(Long.parseLong(string)),
                position -> Long.toString(position.value)
        );

        private final long value;

        private TimePosition(long value) {
{'fraction_non_alphanumeric': 0.09282488710486704, 'fraction_numerical': 0.0016725204883759827, 'mean_word_length': 2.7398373983739837, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1695155', 'n_tokens_mistral': 1583, 'n_tokens_neox': 1542, 'n_words': 356}