<commit_before>package eventsrc4j.memory;

import eventsrc4j.Event;
import eventsrc4j.Events;
import eventsrc4j.Sequence;
import eventsrc4j.StreamReader;
import eventsrc4j.WriteResult;
import eventsrc4j.WriteResults;
import eventsrc4j.io.EventStorage;
import eventsrc4j.io.EventStream;
import eventsrc4j.io.IO;
import eventsrc4j.io.WritableEventStream;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import java.util.stream.Stream;

import static eventsrc4j.WriteResults.Success;
import static eventsrc4j.memory.GlobalSeqs.seq;
import static eventsrc4j.util.Streams.dropWhile;
import static eventsrc4j.util.Streams.takeWhile;
import static java.util.Collections.emptyList;
import static java.util.Collections.unmodifiableList;
import static java.util.Optional.ofNullable;

public final class MemoryEventStorage<K, S, E> implements EventStorage<K, S, E>, EventStream<K, GlobalSeq<S>, E> {

  private final ConcurrentSkipListMap<S, List<Event<K, S, E>>> globalStream = new ConcurrentSkipListMap<>();

  private final ConcurrentMap<K, WritableEventStream<K, S, E>> streams = new ConcurrentHashMap<>();

  private final Sequence<S> sequence;

  private final AtomicReference<S> nextGlobalSeq;

  public MemoryEventStorage(Sequence<S> sequence) {
    this.sequence = sequence;
    nextGlobalSeq = new AtomicReference<>(sequence.first());
  }


  @Override
  public WritableEventStream<K, S, E> stream(K key) {
    return streams.computeIfAbsent(key, MemoryWritableEventStream::new);
  }

  @Override
  public <R> IO<R> read(Optional<GlobalSeq<S>> fromSeq,
      StreamReader<K, GlobalSeq<S>, E, R> streamReader) {

    return () -> streamReader.apply(
        takeWhile(
            // We limit the stream to consecutive sequence otherwise we could miss some not yet inserted events
            // (because global sequence number are pre-allocated)
            new Predicate<Event<K, GlobalSeq<S>, E>>() {

              S expectedGlobalSeq = fromSeq.map(GlobalSeqs::getGlobalSeq).orElse(sequence.first());

              @Override public boolean test(Event<K, GlobalSeq<S>, E> e) {
                if (e.seq().globalSeq().equals(expectedGlobalSeq)) {
                  expectedGlobalSeq = sequence.next(expectedGlobalSeq);
                  return true;
                }
                return false;
              }
            }
            ,
            fromSeq.map(fromSeqExlusive ->
                dropWhile(
                    (Event<K, GlobalSeq<S>, E> e) ->
                        e.seq().globalSeq().equals(fromSeqExlusive.globalSeq())
                            && sequence.compare(e.seq().seq(), fromSeqExlusive.seq()) <= 0,

                    flattenGlobalStream(globalStream.tailMap(fromSeqExlusive.globalSeq()))
                )
            ).orElse(flattenGlobalStream(globalStream))
        )
    );
  }

  @Override
  public IO<Optional<Event<K, GlobalSeq<S>, E>>> latest() {
    return () -> ofNullable(globalStream.lastEntry()).map(
        e -> Events.<K, S, E, GlobalSeq<S>>modSeq(s -> seq(e.getKey(), s))
            .apply(e.getValue().get(e.getValue().size() - 1)));
  }

  private final class MemoryWritableEventStream implements WritableEventStream<K, S, E> {
<commit_msg>MemoryEventStorage: fix bug when reading global stream when there is batch commit of multiple events.<commit_after>package eventsrc4j.memory;

import eventsrc4j.Event;
import eventsrc4j.Events;
import eventsrc4j.Sequence;
import eventsrc4j.StreamReader;
import eventsrc4j.WriteResult;
import eventsrc4j.WriteResults;
import eventsrc4j.io.EventStorage;
import eventsrc4j.io.EventStream;
import eventsrc4j.io.IO;
import eventsrc4j.io.WritableEventStream;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import java.util.stream.Stream;

import static eventsrc4j.WriteResults.Success;
import static eventsrc4j.memory.GlobalSeqs.seq;
import static eventsrc4j.util.Streams.dropWhile;
import static eventsrc4j.util.Streams.takeWhile;
import static java.util.Collections.emptyList;
import static java.util.Collections.unmodifiableList;
import static java.util.Optional.ofNullable;

public final class MemoryEventStorage<K, S, E> implements EventStorage<K, S, E>, EventStream<K, GlobalSeq<S>, E> {

  private final ConcurrentSkipListMap<S, List<Event<K, S, E>>> globalStream = new ConcurrentSkipListMap<>();

  private final ConcurrentMap<K, WritableEventStream<K, S, E>> streams = new ConcurrentHashMap<>();

  private final Sequence<S> sequence;

  private final AtomicReference<S> nextGlobalSeq;

  public MemoryEventStorage(Sequence<S> sequence) {
    this.sequence = sequence;
    nextGlobalSeq = new AtomicReference<>(sequence.first());
  }


  @Override
  public WritableEventStream<K, S, E> stream(K key) {
    return streams.computeIfAbsent(key, MemoryWritableEventStream::new);
  }

  @Override
  public <R> IO<R> read(Optional<GlobalSeq<S>> fromSeq,
      StreamReader<K, GlobalSeq<S>, E, R> streamReader) {

    return () -> streamReader.apply(
        takeWhile(
            // We limit the stream to consecutive sequence otherwise we could miss some not yet inserted events
            // (because global sequence number are pre-allocated)
            new Predicate<Event<K, GlobalSeq<S>, E>>() {

              S nextExpectedGlobalSeq = fromSeq.map(GlobalSeqs::getGlobalSeq).orElse(sequence.first());

              @Override public boolean test(Event<K, GlobalSeq<S>, E> e) {
                if (e.seq().globalSeq().equals(nextExpectedGlobalSeq)) {
                  nextExpectedGlobalSeq = sequence.next(nextExpectedGlobalSeq);
                  return true;
                }
                return sequence.compare(e.seq().globalSeq(), nextExpectedGlobalSeq) < 0;
              }
            }
            ,
            fromSeq.map(fromSeqExlusive ->
                dropWhile(
                    (Event<K, GlobalSeq<S>, E> e) ->
                        e.seq().globalSeq().equals(fromSeqExlusive.globalSeq())
                            && sequence.compare(e.seq().seq(), fromSeqExlusive.seq()) <= 0,

                    flattenGlobalStream(globalStream.tailMap(fromSeqExlusive.globalSeq()))
                )
            ).orElse(flattenGlobalStream(globalStream))
        )
    );
  }

  @Override
  public IO<Optional<Event<K, GlobalSeq<S>, E>>> latest() {
    return () -> ofNullable(globalStream.lastEntry()).map(
        e -> Events.<K, S, E, GlobalSeq<S>>modSeq(s -> seq(e.getKey(), s))
            .apply(e.getValue().get(e.getValue().size() - 1)));
  }

  private final class MemoryWritableEventStream implements WritableEventStream<K, S, E> {

{'fraction_non_alphanumeric': 0.11371374097780199, 'fraction_numerical': 0.004766444232602479, 'mean_word_length': 3.4081632653061225, 'pattern_counts': {'":': 0, '<': 78, '<?xml version=': 0, '>': 87, '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': '135916', 'n_tokens_mistral': 2160, 'n_tokens_neox': 2104, 'n_words': 458}