package io.activej.datastream.consumer;

import io.activej.common.ref.RefInt;
import io.activej.datastream.supplier.AbstractStreamSupplier;
import io.activej.datastream.supplier.StreamSuppliers;
import io.activej.promise.Promises;
import io.activej.test.ExpectedException;
import io.activej.test.rules.EventloopRule;
import org.junit.ClassRule;
import org.junit.Test;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.stream.IntStream;

import static io.activej.datastream.TestStreamTransformers.randomlySuspending;
import static io.activej.datastream.TestUtils.assertClosedWithError;
import static io.activej.datastream.TestUtils.assertEndOfStream;
import static io.activej.promise.TestUtils.await;
import static io.activej.promise.TestUtils.awaitException;
import static org.junit.Assert.*;

public class SwitcherStreamConsumerTest {

	@ClassRule
	public static final EventloopRule eventloopRule = new EventloopRule();

	@Test
	public void testSwitching() {
		List<ToListStreamConsumer<Integer>> consumers = IntStream.range(0, 10)
			.mapToObj($ -> ToListStreamConsumer.<Integer>create())
			.toList();
		SwitcherStreamConsumer<Integer> switcher = SwitcherStreamConsumer.create();

		AbstractStreamSupplier<Integer> streamSupplier = new AbstractStreamSupplier<>() {
			final RefInt refInt = new RefInt(0);
			final Iterator<ToListStreamConsumer<Integer>> iterator = consumers.iterator();

			@Override
			protected void onStarted() {
				switcher.switchTo(iterator.next().transformWith(randomlySuspending()));
			}

			@Override
			protected void onResumed() {
				while (isReady()) {
					send(refInt.inc());
					if (refInt.get() == 10) {
						if (iterator.hasNext()) {
							refInt.set(0);
							switcher.switchTo(iterator.next().transformWith(randomlySuspending()));
						} else {
							break;
						}
					}
				}
				if (!iterator.hasNext() && refInt.get() == 10) {
					sendEndOfStream();
				}
			}
		};

		await(streamSupplier.streamTo(switcher));

		assertEndOfStream(streamSupplier, switcher);
		consumers.forEach(consumer -> {
			assertEquals(List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), consumer.getList());
			assertEndOfStream(consumer);
		});
	}

	@Test
	public void testSwitchingToClosedStream() {
		ExpectedException expectedException = new ExpectedException();

		SwitcherStreamConsumer<Integer> switcher = SwitcherStreamConsumer.create();
		ToListStreamConsumer<Integer> consumer1 = ToListStreamConsumer.create();
		StreamConsumer<Integer> consumerClosed = StreamConsumers.closingWithError(expectedException);
		ToListStreamConsumer<Integer> consumer2 = ToListStreamConsumer.create();

		AbstractStreamSupplier<Integer> streamSupplier = new AbstractStreamSupplier<>() {
			final RefInt refInt = new RefInt(0);

			@Override
			protected void onStarted() {
				switcher.switchTo(consumer1.transformWith(randomlySuspending()));
			}

			@Override
			protected void onResumed() {
				while (isReady()) {
					send(refInt.inc());
					int number = refInt.get();
					if (number % 10 == 0) {
						if (number == 10) {
							switcher.switchTo(consumerClosed.transformWith(randomlySuspending()));
						} else if (number == 20) {
							switcher.switchTo(consumer2.transformWith(randomlySuspending()));
						} else {
							break;
						}
					}
				}
				if (refInt.get() == 30) {
					sendEndOfStream();
				}
			}
		};

		Exception exception = awaitException(streamSupplier.streamTo(switcher));

		assertSame(expectedException, exception);
		assertClosedWithError(expectedException, streamSupplier, switcher);

//		assertClosedWithError(consumer1);
		assertEquals(List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), consumer1.getList());

		assertClosedWithError(expectedException, consumerClosed);
		assertFalse(consumer2.isStarted());
	}

	@Test
	public void testSwitchingAfterSwitcherIsDone() {
		SwitcherStreamConsumer<Integer> switcher = SwitcherStreamConsumer.create();
		ToListStreamConsumer<Integer> consumer = ToListStreamConsumer.create();

		switcher.switchTo(consumer);

		await(StreamSuppliers.ofValues(1, 2, 3, 4).streamTo(switcher));

		assertEndOfStream(switcher);
		assertEndOfStream(consumer);

		assertEquals(List.of(1, 2, 3, 4), consumer.getList());
	}

	@Test
	public void testSwitchingAfterSwitcherIsClosed() {
		ExpectedException expectedException = new ExpectedException();

		SwitcherStreamConsumer<Integer> switcher = SwitcherStreamConsumer.create();
		switcher.closeEx(expectedException);

		assertSame(expectedException, awaitException(StreamSuppliers.ofValues(1, 2, 3, 4).streamTo(switcher)));

		assertClosedWithError(expectedException, switcher);
	}

	@Test
	public void testSwitchingBeforeSwitcherIsBound() {
		SwitcherStreamConsumer<Integer> switcher = SwitcherStreamConsumer.create();
		ToListStreamConsumer<Integer> consumer = ToListStreamConsumer.create();

		switcher.switchTo(consumer);

		await(StreamSuppliers.ofValues(1, 2, 3, 4).streamTo(switcher));

		assertEndOfStream(switcher);
		assertEndOfStream(consumer);

		assertEquals(List.of(1, 2, 3, 4), consumer.getList());
	}

	@Test
	public void testSwitchingToSlowConsumer() {
		ArrayList<Integer> list1 = new ArrayList<>();
		ArrayList<Integer> list2 = new ArrayList<>();
		List<StreamConsumer<Integer>> consumers = List.of(ToListStreamConsumer.create(list1),
			StreamConsumers.ofPromise(Promises.delay(Duration.ofMillis(1), ToListStreamConsumer.create(list2))));
		SwitcherStreamConsumer<Integer> switcher = SwitcherStreamConsumer.create();

		AbstractStreamSupplier<Integer> streamSupplier = new AbstractStreamSupplier<>() {
			final RefInt refInt = new RefInt(0);
			final Iterator<StreamConsumer<Integer>> iterator = consumers.iterator();

			@Override
			protected void onStarted() {
				switcher.switchTo(iterator.next());
			}

			@Override
			protected void onResumed() {
				while (isReady()) {
					send(refInt.inc());
					if (refInt.get() == 10) {
						if (iterator.hasNext()) {
							refInt.set(0);
							switcher.switchTo(iterator.next());
						} else {
							break;
						}
					}
				}
				if (!iterator.hasNext() && refInt.get() == 10) {
					sendEndOfStream();
				}
			}
		};

		await(streamSupplier.streamTo(switcher));

		assertEndOfStream(streamSupplier, switcher);

		List<Integer> expected = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
		assertEquals(expected, list1);
		assertEquals(expected, list2);
	}
}
