package com.timgroup.eventstore.merging;

import com.timgroup.eventstore.api.EventReader;
import com.timgroup.eventstore.api.EventRecord;
import com.timgroup.eventstore.api.Position;
import com.timgroup.eventstore.api.ResolvedEvent;
import com.timgroup.eventstore.api.StreamId;

import java.util.Optional;
import java.util.Spliterator;
import java.util.function.Consumer;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import static com.timgroup.eventstore.api.EventRecord.eventRecord;
import static java.lang.Long.MAX_VALUE;
import static java.util.stream.Collectors.toList;

final class MergedEventReader<T extends Comparable<T>> implements EventReader {

    private final MergingStrategy<T> mergingStrategy;
    private final EventReader[] readers;

    public MergedEventReader(MergingStrategy<T> mergingStrategy, EventReader... readers) {
        this.mergingStrategy = mergingStrategy;
        this.readers = readers;
    }

    @Override
    public Stream<ResolvedEvent> readAllForwards(Position positionExclusive) {
        return StreamSupport.stream(new MergingSpliterator<>(mergingStrategy, (MergedEventReaderPosition) positionExclusive, readers), false);
    }

    private static final class MergingSpliterator<T extends Comparable<T>> implements Spliterator<ResolvedEvent> {

        private final MergingStrategy<T> mergingStrategy;
        private final EventReader[] readers;

        private MergedEventReaderPosition currentPosition;

        private MergingSpliterator(MergingStrategy<T> mergingStrategy, MergedEventReaderPosition startPosition, EventReader... readers) {
            this.mergingStrategy = mergingStrategy;
            this.readers = readers;
            this.currentPosition = startPosition;
        }

        @Override
        public boolean tryAdvance(Consumer<? super ResolvedEvent> consumer) {
            Optional<ResolvedEvent> maybeNextInputEvent = advanceToNextEvent();
            if (maybeNextInputEvent.isPresent()) {
                EventRecord record = maybeNextInputEvent.get().eventRecord();
                consumer.accept(new ResolvedEvent(
                        currentPosition,
                        eventRecord(
                                record.timestamp(),
                                StreamId.streamId("input", "all"),
                                currentPosition.outputEventNumber,
                                record.eventType(),
                                record.data(),
                                record.metadata()
                        )
                ));
                return true;
            }
            return false;
        }

        private Optional<ResolvedEvent> advanceToNextEvent() {
            Optional<T> candidateOrderingValue = Optional.empty();
            Optional<ResolvedEvent> candidate = Optional.empty();
            int candidateIndex = -1;

            for (int readerIndex = 0; readerIndex < readers.length; readerIndex++) {
                EventReader reader = readers[readerIndex];
                Optional<ResolvedEvent> maybeEvent = reader.readAllForwards(currentPosition.inputPositions[readerIndex]).findFirst();
                if (maybeEvent.isPresent()) {
                    T orderingValue = mergingStrategy.toComparable(maybeEvent.get());
                    if (!candidateOrderingValue.isPresent() || orderingValue.compareTo(candidateOrderingValue.get()) < 0) {
                        candidateOrderingValue = Optional.of(orderingValue);
                        candidate = maybeEvent;
                        candidateIndex = readerIndex;
                    }
                }
            }

            if (candidate.isPresent()) {
                currentPosition = currentPosition.withNextPosition(candidateIndex, candidate.get().position());
            }
            return candidate;
        }

        @Override
        public Spliterator<ResolvedEvent> trySplit() {
            return null;
        }

        @Override
        public long estimateSize() {
            return MAX_VALUE;
        }

        @Override
        public int characteristics() {
            return ORDERED | NONNULL | DISTINCT;
        }
    }

    @Override
    public Position emptyStorePosition() {
        Position[] positions = Stream.of(readers).map(EventReader::emptyStorePosition).collect(toList()).toArray(new Position[0]);
        return new MergedEventReaderPosition(-1L, positions);
    }

}

@@ -35,16 +35,110 @@
 rging;%0A%0A
+import com.google.common.collect.Iterators;%0Aimport com.google.common.collect.PeekingIterator;%0A
 import c
@@ -366,16 +366,71 @@
 eamId;%0A%0A
+import java.util.Arrays;%0Aimport java.util.Collections;%0A
 import j
@@ -586,16 +586,83 @@
 pport;%0A%0A
+import static com.google.common.collect.Iterators.peekingIterator;%0A
 import s
@@ -760,16 +760,104 @@
 _VALUE;%0A
+import static java.util.Arrays.fill;%0Aimport static java.util.Collections.emptyIterator;%0A
 import s
@@ -1712,24 +1712,94 @@
 %5B%5D readers;%0A
+        private final PeekingIterator%3CResolvedEvent%3E%5B%5D currentBatch;%0A%0A
 %0A        pri
@@ -2065,32 +2065,173 @@
 ders = readers;%0A
+            this.currentBatch = new PeekingIterator%5Breaders.length%5D;%0A            fill(this.currentBatch, peekingIterator(emptyIterator()));%0A%0A
             this
@@ -2267,32 +2267,33 @@
 ion;%0A        %7D%0A%0A
+%0A
         @Overrid
@@ -3606,86 +3606,31 @@
 t = 
-reader.readAllForwards(currentPosition.inputPositions%5BreaderIndex%5D).findFirst(
+peekNextFor(readerIndex
 );%0A 
@@ -4287,38 +4287,620 @@
     
-%7D%0A            return candidate
+    currentBatch%5BcandidateIndex%5D.next();%0A            %7D%0A            return candidate;%0A        %7D%0A%0A        private Optional%3CResolvedEvent%3E peekNextFor(int readerIndex) %7B%0A            Position pos = currentPosition.inputPositions%5BreaderIndex%5D;%0A%0A            if (currentBatch%5BreaderIndex%5D.hasNext()) %7B%0A                return Optional.of(currentBatch%5BreaderIndex%5D.peek());%0A            %7D%0A%0A            currentBatch%5BreaderIndex%5D = peekingIterator(readers%5BreaderIndex%5D.readAllForwards(pos).iterator());%0A            return currentBatch%5BreaderIndex%5D.hasNext() ? Optional.of(currentBatch%5BreaderIndex%5D.peek()) : Optional.empty()
 ;%0A  

reimplement the Merging Spliterator to buffer reads from the underlying readers