package com.timgroup.eventstore.merging;

import com.timgroup.eventstore.api.EventCategoryReader;
import com.timgroup.eventstore.api.EventReader;
import com.timgroup.eventstore.api.EventSource;
import com.timgroup.eventstore.api.Position;
import com.timgroup.eventstore.api.PositionCodec;
import com.timgroup.eventstore.api.ResolvedEvent;

import java.util.function.Function;
import java.util.stream.Stream;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;

public final class NamedReaderWithCodec {
    final String name;
    final Function<? super Position, ? extends Stream<ResolvedEvent>> reader;
    final PositionCodec codec;
    final Position startingPosition;

    public NamedReaderWithCodec(String name, EventReader reader) {
        this(name, reader, reader.storePositionCodec(), reader.emptyStorePosition());
    }

    public NamedReaderWithCodec(String name, EventReader reader, PositionCodec codec) {
        this(name, reader, codec, reader.emptyStorePosition());
    }

    public NamedReaderWithCodec(String name, EventReader reader, Position startingPosition) {
        this(name, reader::readAllForwards, reader.storePositionCodec(), startingPosition);
    }

    public NamedReaderWithCodec(String name, EventReader reader, PositionCodec codec, Position startingPosition) {
        this(name, reader::readAllForwards, codec, startingPosition);
    }

    public NamedReaderWithCodec(String name, Function<? super Position, ? extends Stream<ResolvedEvent>> reader, PositionCodec codec, Position startingPosition) {
        checkNotNull(name, "name cannot be null");

        final String candidateName = name.trim();
        checkArgument(!candidateName.isEmpty(), "name cannot be empty");

        this.name = candidateName.intern();
        this.reader = reader;
        this.codec = codec;
        this.startingPosition = startingPosition;
    }

    /**
     * @deprecated use {@link #fromEventReader(String, EventReader)}
     */
    @Deprecated
    public static NamedReaderWithCodec fromEventSource(String name, EventSource source) {
        return new NamedReaderWithCodec(name, source.readAll(), source.positionCodec());
    }

    public static NamedReaderWithCodec fromEventReader(String name, EventReader eventReader) {
        return new NamedReaderWithCodec(name, eventReader, eventReader.storePositionCodec());
    }

    public static NamedReaderWithCodec fromEventCategoryReader(String name, EventCategoryReader eventCategoryReader, String category) {
        return new NamedReaderWithCodec(name, pos -> eventCategoryReader.readCategoryForwards(category, pos), eventCategoryReader.categoryPositionCodec(category), eventCategoryReader.emptyCategoryPosition(category));
    }

    @Override
    public String toString() {
        return "NamedReaderWithCodec{" +
                "name='" + name + '\'' +
                ", codec=" + codec +
                ", startingPosition=" + startingPosition +
                '}';
    }
}

@@ -2322,32 +2322,496 @@
  eventReader) %7B%0A
+        return fromEventReader(name, eventReader, eventReader.emptyStorePosition());%0A    %7D%0A%0A    public static NamedReaderWithCodec fromEventReader(String name, EventReader eventReader, String startingPosition) %7B%0A        return fromEventReader(name, eventReader, eventReader.storePositionCodec().deserializePosition(startingPosition));%0A    %7D%0A%0A    public static NamedReaderWithCodec fromEventReader(String name, EventReader eventReader, Position startingPosition) %7B%0A
         return n
@@ -2877,32 +2877,50 @@
 ePositionCodec()
+, startingPosition
 );%0A    %7D%0A%0A    pu
@@ -3068,102 +3068,561 @@
 urn 
-new NamedReaderWithCodec(name, pos -%3E eventCategoryReader.readCategoryForwards(category, pos),
+fromEventCategoryReader(name, eventCategoryReader, category, eventCategoryReader.emptyCategoryPosition(category));%0A    %7D%0A%0A    public static NamedReaderWithCodec fromEventCategoryReader(String name, EventCategoryReader eventCategoryReader, String category, String startingPosition) %7B%0A        return fromEventCategoryReader(name, eventCategoryReader, category, eventCategoryReader.categoryPositionCodec(category).deserializePosition(startingPosition));%0A    %7D%0A%0A    public static NamedReaderWithCodec fromEventCategoryReader(String name, EventCategoryReader
  eve
@@ -3633,33 +3633,41 @@
 tegoryReader
-.
+, String 
 category
 PositionCode
@@ -3650,32 +3650,34 @@
 ing category
+, 
 Position
 Codec(catego
@@ -3656,37 +3656,146 @@
 tegory, Position
-Codec
+ startingPosition) %7B%0A        return new NamedReaderWithCodec(name, pos -%3E eventCategoryReader.readCategoryForwards
 (category), even
@@ -3787,16 +3787,21 @@
 category
+, pos
 ), event
@@ -3807,38 +3807,33 @@
 tCategoryReader.
-emptyC
+c
 ategoryPosition(
@@ -3823,32 +3823,37 @@
 categoryPosition
+Codec
 (category));%0A   
@@ -3838,32 +3838,50 @@
 nCodec(category)
+, startingPosition
 );%0A    %7D%0A%0A    @O

support creating NamedReaderWithCodec with starting positions