<commit_before>package org.elder.sourcerer.esjc;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.msemys.esjc.CannotEstablishConnectionException;
import com.github.msemys.esjc.CatchUpSubscription;
import com.github.msemys.esjc.CatchUpSubscriptionListener;
import com.github.msemys.esjc.CatchUpSubscriptionSettings;
import com.github.msemys.esjc.ConnectionClosedException;
import com.github.msemys.esjc.EventStore;
import com.github.msemys.esjc.EventStoreException;
import com.github.msemys.esjc.ResolvedEvent;
import com.github.msemys.esjc.SliceReadStatus;
import com.github.msemys.esjc.StreamEventsSlice;
import com.github.msemys.esjc.SubscriptionDropReason;
import com.github.msemys.esjc.WriteResult;
import com.github.msemys.esjc.node.cluster.ClusterException;
import com.github.msemys.esjc.operation.AccessDeniedException;
import com.github.msemys.esjc.operation.CommandNotExpectedException;
import com.github.msemys.esjc.operation.InvalidTransactionException;
import com.github.msemys.esjc.operation.NoResultException;
import com.github.msemys.esjc.operation.NotAuthenticatedException;
import com.github.msemys.esjc.operation.ServerErrorException;
import com.github.msemys.esjc.operation.StreamDeletedException;
import com.github.msemys.esjc.operation.WrongExpectedVersionException;
import com.github.msemys.esjc.operation.manager.OperationTimeoutException;
import com.github.msemys.esjc.operation.manager.RetriesLimitReachedException;
import com.github.msemys.esjc.subscription.MaximumSubscribersReachedException;
import com.github.msemys.esjc.subscription.PersistentSubscriptionDeletedException;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import org.elder.sourcerer.EventData;
import org.elder.sourcerer.EventNormalizer;
import org.elder.sourcerer.EventReadResult;
import org.elder.sourcerer.EventRecord;
import org.elder.sourcerer.EventRepository;
import org.elder.sourcerer.EventSubscriptionUpdate;
import org.elder.sourcerer.ExpectedVersion;
import org.elder.sourcerer.exceptions.PermanentEventReadException;
import org.elder.sourcerer.exceptions.PermanentEventWriteException;
import org.elder.sourcerer.exceptions.RetriableEventReadException;
import org.elder.sourcerer.exceptions.RetriableEventWriteException;
import org.elder.sourcerer.exceptions.UnexpectedVersionException;
import org.elder.sourcerer.utils.ElderPreconditions;
import org.elder.sourcerer.utils.ImmutableListCollector;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;

/**
 * Sourcerer event repository implementation using EventStore (geteventstore.com) as the underlying
 * system. The EventStore implementation uses Jackson to serialize and deserialize events that are
 * subclasses of the given event type base class. To instruct Jackson on how to correctly
 * deserialize events to the correct concrete sub type, please use either the Jackson annotations,
 * or JAXB annotations as per
 * <a href="http://wiki.fasterxml.com/JacksonPolymorphicDeserialization">Jackson Polymorphic
 * Deserialization</a>
 *
 * @param <T> The type of events managed by the event repository.
 */
public class EventStoreEsjcEventRepository<T> implements EventRepository<T> {
    private static final Logger logger
            = LoggerFactory.getLogger(EventStoreEsjcEventRepository.class);

    private static final int MAX_MAX_EVENTS_PER_READ = 4095;
    private static final long DEFAULT_TIMEOUT_MILLIS = 30 * (long) 1000;
    private final String streamPrefix;
    private final Class<T> eventClass;
    private final EventStore eventStore;
    private final ObjectMapper objectMapper;
    private final EventNormalizer<T> normalizer;
    private final long timeoutMillis;
    private final CatchUpSubscriptionSettings defaultSubscriptionSettings;

    public EventStoreEsjcEventRepository(
            final String streamPrefix,
            final EventStore eventStore,
            final Class<T> eventClass,
            final ObjectMapper objectMapper,
            final EventNormalizer<T> normalizer) {
        this.streamPrefix = streamPrefix;
        this.eventClass = eventClass;
        this.eventStore = eventStore;
        this.objectMapper = objectMapper;
        this.normalizer = normalizer;
        this.timeoutMillis = DEFAULT_TIMEOUT_MILLIS;
        // TODO: Customize these settings
        defaultSubscriptionSettings = CatchUpSubscriptionSettings.newBuilder()
                .resolveLinkTos(true)
                .build();
    }

    @Override
    public Class<T> getEventType() {
        return eventClass;
    }

    @Override
    public EventReadResult<T> readAll(final int version, final int maxEvents) {
        return readInternal(getCategoryStreamName(), version, maxEvents, true);
    }

    @Override
    public EventReadResult<T> read(final String streamId, final int version, final int maxEvents) {
        return readInternal(toEsStreamId(streamId), version, maxEvents, false);
    }

    @Override
    public EventRecord<T> readFirst(final String streamId) {
        return readSingleInternal(toEsStreamId(streamId), 0, false);
    }

    @Override
    public EventRecord<T> readLast(final String streamId) {
        return readSingleInternal(toEsStreamId(streamId), -1, false);
    }

    // Not part of public interface, but used for inspection / testing
    private String getCategoryStreamName() {
        return "$ce-" + streamPrefix;
    }<commit_msg>Cleanup<commit_after>package org.elder.sourcerer.esjc;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.msemys.esjc.CannotEstablishConnectionException;
import com.github.msemys.esjc.CatchUpSubscription;
import com.github.msemys.esjc.CatchUpSubscriptionListener;
import com.github.msemys.esjc.CatchUpSubscriptionSettings;
import com.github.msemys.esjc.ConnectionClosedException;
import com.github.msemys.esjc.EventStore;
import com.github.msemys.esjc.EventStoreException;
import com.github.msemys.esjc.ResolvedEvent;
import com.github.msemys.esjc.SliceReadStatus;
import com.github.msemys.esjc.StreamEventsSlice;
import com.github.msemys.esjc.SubscriptionDropReason;
import com.github.msemys.esjc.WriteResult;
import com.github.msemys.esjc.node.cluster.ClusterException;
import com.github.msemys.esjc.operation.AccessDeniedException;
import com.github.msemys.esjc.operation.CommandNotExpectedException;
import com.github.msemys.esjc.operation.InvalidTransactionException;
import com.github.msemys.esjc.operation.NoResultException;
import com.github.msemys.esjc.operation.NotAuthenticatedException;
import com.github.msemys.esjc.operation.ServerErrorException;
import com.github.msemys.esjc.operation.StreamDeletedException;
import com.github.msemys.esjc.operation.WrongExpectedVersionException;
import com.github.msemys.esjc.operation.manager.OperationTimeoutException;
import com.github.msemys.esjc.operation.manager.RetriesLimitReachedException;
import com.github.msemys.esjc.subscription.MaximumSubscribersReachedException;
import com.github.msemys.esjc.subscription.PersistentSubscriptionDeletedException;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import org.elder.sourcerer.EventData;
import org.elder.sourcerer.EventNormalizer;
import org.elder.sourcerer.EventReadResult;
import org.elder.sourcerer.EventRecord;
import org.elder.sourcerer.EventRepository;
import org.elder.sourcerer.EventSubscriptionUpdate;
import org.elder.sourcerer.ExpectedVersion;
import org.elder.sourcerer.exceptions.PermanentEventReadException;
import org.elder.sourcerer.exceptions.PermanentEventWriteException;
import org.elder.sourcerer.exceptions.RetriableEventReadException;
import org.elder.sourcerer.exceptions.RetriableEventWriteException;
import org.elder.sourcerer.exceptions.UnexpectedVersionException;
import org.elder.sourcerer.utils.ElderPreconditions;
import org.elder.sourcerer.utils.ImmutableListCollector;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;

/**
 * Sourcerer event repository implementation using EventStore (geteventstore.com) as the underlying
 * system. The EventStore implementation uses Jackson to serialize and deserialize events that are
 * subclasses of the given event type base class. To instruct Jackson on how to correctly
 * deserialize events to the correct concrete sub type, please use either the Jackson annotations,
 * or JAXB annotations as per
 * <a href="http://wiki.fasterxml.com/JacksonPolymorphicDeserialization">Jackson Polymorphic
 * Deserialization</a>
 *
 * @param <T> The type of events managed by the event repository.
 */
public class EventStoreEsjcEventRepository<T> implements EventRepository<T> {
    private static final Logger logger
            = LoggerFactory.getLogger(EventStoreEsjcEventRepository.class);

    private static final int MAX_MAX_EVENTS_PER_READ = 4095;
    private static final long DEFAULT_TIMEOUT_MILLIS = 30 * (long) 1000;
    private final String streamPrefix;
    private final Class<T> eventClass;
    private final EventStore eventStore;
    private final ObjectMapper objectMapper;
    private final EventNormalizer<T> normalizer;
    private final long timeoutMillis;
    private final CatchUpSubscriptionSettings defaultSubscriptionSettings;

    public EventStoreEsjcEventRepository(
            final String streamPrefix,
            final EventStore eventStore,
            final Class<T> eventClass,
            final ObjectMapper objectMapper,
            final EventNormalizer<T> normalizer) {
        this.streamPrefix = streamPrefix;
        this.eventClass = eventClass;
        this.eventStore = eventStore;
        this.objectMapper = objectMapper;
        this.normalizer = normalizer;
        this.timeoutMillis = DEFAULT_TIMEOUT_MILLIS;
        // TODO: Customize these settings
        defaultSubscriptionSettings = CatchUpSubscriptionSettings.newBuilder()
                .resolveLinkTos(true)
                .build();
    }

    @Override
    public Class<T> getEventType() {
        return eventClass;
    }

    @Override
    public EventReadResult<T> readAll(final int version, final int maxEvents) {
        return readInternal(getCategoryStreamName(), version, maxEvents, true);
    }

    @Override
    public EventReadResult<T> read(final String streamId, final int version, final int maxEvents) {
        return readInternal(toEsStreamId(streamId), version, maxEvents, false);
    }

    @Override
    public EventRecord<T> readFirst(final String streamId) {
        return readSingleInternal(toEsStreamId(streamId), 0, false);
    }

    @Override
    public EventRecord<T> readLast(final String streamId) {
        return readSingleInternal(toEsStreamId(streamId), -1, false);
    }

    private String getCategoryStreamName() {
        return "$ce-" + streamPrefix;
    }
{'fraction_non_alphanumeric': 0.08085430968726164, 'fraction_numerical': 0.0023730824646156453, 'mean_word_length': 4.953582240161453, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, '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': '5940403', 'n_tokens_mistral': 3493, 'n_tokens_neox': 3315, 'n_words': 692}