package com.timgroup.eventsubscription;

import com.timgroup.eventstore.api.EventInStream;
import com.timgroup.eventstore.api.EventReader;
import com.timgroup.eventstore.api.EventStore;
import com.timgroup.eventstore.api.EventStreamReader;
import com.timgroup.eventstore.api.LegacyPositionAdapter;
import com.timgroup.eventstore.api.Position;
import com.timgroup.eventstore.api.ResolvedEvent;
import com.timgroup.eventstore.api.StreamId;

import java.time.Instant;
import java.util.stream.Stream;

import static com.timgroup.eventstore.api.EventRecord.eventRecord;
import static com.timgroup.eventstore.api.StreamId.streamId;

public class LegacyEventStoreEventReaderAdapter implements EventReader, EventStreamReader {
    private final EventStore eventStore;

    public LegacyEventStoreEventReaderAdapter(EventStore eventStore) {
        this.eventStore = eventStore;
    }

    @Override
    public Stream<ResolvedEvent> readAllForwards() {
        return eventStore.streamingFromAll(0).map(this::toResolvedEvent);
    }

    @Override
    public Stream<ResolvedEvent> readAllForwards(Position positionExclusive) {
        return eventStore.streamingFromAll(((LegacyPositionAdapter) positionExclusive).version()).map(this::toResolvedEvent);
    }

    @Override
    public Position emptyStorePosition() {
        return new LegacyPositionAdapter(0);
    }

    private ResolvedEvent toResolvedEvent(EventInStream eventInStream) {
        return new ResolvedEvent(
                new LegacyPositionAdapter(eventInStream.version()),
                eventRecord(
                        Instant.ofEpochMilli(eventInStream.effectiveTimestamp().getMillis()),
                        streamId("all", "all"),
                        eventInStream.version(),
                        eventInStream.eventData().eventType(),
                        eventInStream.eventData().body().data(),
                        new byte[0]
                )
        );
    }

    @Override
    public Stream<ResolvedEvent> readStreamForwards(StreamId streamId, long eventNumber) {
        if (!streamId.equals(streamId("all", "all"))) {
            throw new IllegalArgumentException("Cannot read " + streamId + " using legacy adapter");
        }
        return eventStore.streamingFromAll(eventNumber).map(this::toResolvedEvent);
    }
}

@@ -179,16 +179,64 @@
 tStore;%0A
+import com.timgroup.eventstore.api.EventStream;%0A
 import c
@@ -273,32 +273,32 @@
 ntStreamReader;%0A
-
 import com.timgr
@@ -502,24 +502,51 @@
 me.Instant;%0A
+import java.util.Iterator;%0A
 import java.
@@ -564,16 +564,55 @@
 .Stream;
+%0Aimport java.util.stream.StreamSupport;
 %0A%0Aimport
@@ -732,16 +732,193 @@
 treamId;
+%0Aimport static java.util.Spliterator.ORDERED;%0Aimport static java.util.Spliterators.spliteratorUnknownSize;%0Aimport static scala.collection.JavaConverters.asJavaIteratorConverter;
 %0A%0Apublic
@@ -1398,24 +1398,24 @@
 xclusive) %7B%0A
-
         retu
@@ -1527,32 +1527,436 @@
 dEvent);%0A    %7D%0A%0A
+    public Stream%3CResolvedEvent%3E readAllForwardsBuffered(Position positionExclusive) %7B%0A        EventStream eventStream = eventStore.fromAll(((LegacyPositionAdapter) positionExclusive).version());%0A        Iterator%3CEventInStream%3E events = asJavaIteratorConverter(eventStream).asJava();%0A%0A        return StreamSupport.stream(spliteratorUnknownSize(events, ORDERED), false).map(this::toResolvedEvent);%0A    %7D%0A%0A
     @Override%0A  

add a method to allow buffered buttered reading through the legacy adapter