package org.zalando.nakadi.service.subscription.state;

import org.junit.BeforeClass;
import org.junit.Test;
import org.zalando.nakadi.domain.ConsumedEvent;
import org.zalando.nakadi.domain.NakadiCursor;
import org.zalando.nakadi.domain.Storage;
import org.zalando.nakadi.domain.Timeline;
import org.zalando.nakadi.repository.kafka.KafkaCursor;

import java.util.Comparator;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static java.lang.System.currentTimeMillis;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

public class PartitionDataTest {

    private static Timeline firstTimeline = mock(Timeline.class);
    private static final Comparator<NakadiCursor> COMP = Comparator.comparing(NakadiCursor::getOffset);

    @BeforeClass
    public static void initTimeline() {
        when(firstTimeline.getStorage()).thenReturn(new Storage("", Storage.Type.KAFKA));
    }


    private static NakadiCursor createCursor(final long offset) {
        return new KafkaCursor("x", 0, offset).toNakadiCursor(firstTimeline);
    }

    @Test
    public void onNewOffsetsShouldSupportRollback() {
        final PartitionData pd = new PartitionData(COMP, null, createCursor(100L), System.currentTimeMillis());
        final PartitionData.CommitResult cr = pd.onCommitOffset(createCursor(90L));

        assertEquals(0L, cr.committedCount);
        assertEquals(true, cr.seekOnKafka);
        assertEquals(90L, Long.parseLong(pd.getSentOffset().getOffset()));
        assertEquals(0L, pd.getUnconfirmed());
    }

    @Test
    public void onNewOffsetsShouldSupportCommitInFuture() {
        final PartitionData pd = new PartitionData(COMP, null, createCursor(100L), System.currentTimeMillis());
        final PartitionData.CommitResult cr = pd.onCommitOffset(createCursor(110L));

        assertEquals(0L, cr.committedCount);
        assertEquals(true, cr.seekOnKafka);
        assertEquals(110L, Long.parseLong(pd.getSentOffset().getOffset()));
        assertEquals(0L, pd.getUnconfirmed());
    }

    @Test
    public void normalOperationShouldNotReconfigureKafkaConsumer() {
        final PartitionData pd = new PartitionData(COMP, null, createCursor(100L), System.currentTimeMillis());
        for (long i = 0; i < 100; ++i) {
            pd.addEvent(new ConsumedEvent(("test_" + i).getBytes(), createCursor(100L + i + 1)));
        }
        // Now say to it that it was sent
        pd.takeEventsToStream(currentTimeMillis(), 1000, 0L);
        assertEquals(100L, pd.getUnconfirmed());
        for (long i = 0; i < 10; ++i) {
            final PartitionData.CommitResult cr = pd.onCommitOffset(createCursor(110L + i * 10L));
            assertEquals(10L, cr.committedCount);
            assertFalse(cr.seekOnKafka);
            assertEquals(90L - i * 10L, pd.getUnconfirmed());
        }
    }

    @Test
    public void keepAliveCountShouldIncreaseOnEachEmptyCall() {
        final PartitionData pd = new PartitionData(COMP, null, createCursor(100L), System.currentTimeMillis());
        for (int i = 0; i < 100; ++i) {
            pd.takeEventsToStream(currentTimeMillis(), 10, 0L);
            assertEquals(i + 1, pd.getKeepAliveInARow());
        }
        pd.addEvent(new ConsumedEvent("".getBytes(), createCursor(101L)));
        assertEquals(100, pd.getKeepAliveInARow());
        pd.takeEventsToStream(currentTimeMillis(), 10, 0L);
        assertEquals(0, pd.getKeepAliveInARow());
        pd.takeEventsToStream(currentTimeMillis(), 10, 0L);
        assertEquals(1, pd.getKeepAliveInARow());
    }

    @Test
    public void eventsShouldBeStreamedOnTimeout() throws InterruptedException {
        final long timeout = TimeUnit.SECONDS.toMillis(1);
        long currentTime = System.currentTimeMillis();

        final PartitionData pd = new PartitionData(COMP, null, createCursor(100L), currentTime);
        for (int i = 0; i < 100; ++i) {
            pd.addEvent(new ConsumedEvent("test".getBytes(), createCursor(i + 100L + 1)));
        }
        List<ConsumedEvent> data = pd.takeEventsToStream(currentTime, 1000, timeout);
        assertNull(data);
        assertEquals(0, pd.getKeepAliveInARow());

        currentTime += timeout + 1;

        data = pd.takeEventsToStream(currentTime, 1000, timeout);
        assertNotNull(data);
        assertEquals(100, data.size());

        for (int i = 100; i < 200; ++i) {
            pd.addEvent(new ConsumedEvent("test".getBytes(), createCursor(i + 100L + 1)));
        }
        data = pd.takeEventsToStream(currentTime, 1000, timeout);
        assertNull(data);
        assertEquals(0, pd.getKeepAliveInARow());

        currentTime += timeout + 1;

        data = pd.takeEventsToStream(currentTime, 1000, timeout);
        assertNotNull(data);
        assertEquals(100, data.size());
    }

    @Test
    public void eventsShouldBeStreamedOnBatchSize() {
        final long timeout = TimeUnit.SECONDS.toMillis(1);
        final PartitionData pd = new PartitionData(COMP, null, createCursor(100L), System.currentTimeMillis());
        for (int i = 0; i < 100; ++i) {
            pd.addEvent(new ConsumedEvent("test".getBytes(), createCursor(i + 100L + 1)));
        }
        assertNull(pd.takeEventsToStream(currentTimeMillis(), 1000, timeout));
        final List<ConsumedEvent> eventsToStream = pd.takeEventsToStream(currentTimeMillis(), 99, timeout);
        assertNotNull(eventsToStream);
        assertEquals(99, eventsToStream.size());
    }
}

@@ -1098,17 +1098,16 @@
 %0A    %7D%0A%0A
-%0A
     priv
@@ -3792,36 +3792,8 @@
 ut()
- throws InterruptedException
  %7B%0A 

ARUHA-1397 Fix whitespace