package com.github.blahord.integral.job;

import com.github.blahord.integral.ComponentException;
import com.github.blahord.integral.component.ConsumerComponent;
import com.github.blahord.integral.component.ProducerComponent;
import com.github.blahord.integral.data.DataCell;
import com.github.blahord.integral.data.DataRow;
import com.github.blahord.integral.schema.Schema;
import com.github.blahord.integral.schema.SchemaColumn;
import com.github.blahord.integral.type.IntegerType;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import static com.github.blahord.integral.type.DefaultTypes.SIMPLE_INTEGER_TYPE;
import static com.github.blahord.integral.type.DefaultTypes.SIMPLE_NULLABLE_INTEGER_TYPE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

public class IntegralJobTest {

    private Producer producer;
    private Consumer consumer;

    private IntegralJob integralJob;

    @BeforeEach
    public void setUp() {
        producer = new Producer("producer", "out");
        consumer = new Consumer("consumer", "in", true);

        integralJob = new IntegralJob();
    }

    @Test
    public void testAddComponent() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);
    }

    @Test
    public void testAddComponent_DoubleComponent() {
        integralJob.addComponent(producer);
        assertThatThrownBy(() -> integralJob.addComponent(producer)).isInstanceOf(IllegalStateException.class);
    }

    @Test
    public void testConnect() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);

        integralJob.connect(producer, "out", consumer, "in");
    }

    @Test
    public void testConnect_unknownOutComponent() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);

        final Producer prod = new Producer("prod", "out");

        assertThatThrownBy(() -> integralJob.connect(prod, "out", consumer, "in")).isInstanceOf(IllegalArgumentException.class);
    }

    @Test
    public void testConnect_unknownInComponent() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);

        final Consumer cons = new Consumer("cons", "in", true);

        assertThatThrownBy(() -> integralJob.connect(producer, "out", cons, "in")).isInstanceOf(IllegalArgumentException.class);
    }

    @Test
    public void testConnect_unknownOutput() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);

        assertThatThrownBy(() -> integralJob.connect(producer, "out2", consumer, "in")).isInstanceOf(IllegalArgumentException.class);
    }

    @Test
    public void testConnect_unknownInput() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);

        assertThatThrownBy(() -> integralJob.connect(producer, "out", consumer, "in2")).isInstanceOf(IllegalArgumentException.class);
    }

    @Test
    public void testConnect_OutputAlreadyConnected() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);
        final Consumer consumer2 = new Consumer("consumer2", "in", true);
        integralJob.addComponent(consumer2);

        integralJob.connect(producer, "out", consumer, "in");
        assertThatThrownBy(() -> integralJob.connect(producer, "out", consumer2, "in")).isInstanceOf(IllegalStateException.class);
    }

    @Test
    public void testConnect_InputAlreadyConnected() {
        integralJob.addComponent(producer);
        final Producer producer2 = new Producer("producer2", "out");
        integralJob.addComponent(producer2);
        integralJob.addComponent(consumer);

        integralJob.connect(producer, "out", consumer, "in");
        assertThatThrownBy(() -> integralJob.connect(producer2, "out", consumer, "in")).isInstanceOf(IllegalStateException.class);
    }

    @Test
    public void testConnect_incompatibleSchemas() {
        integralJob.addComponent(producer);
        Consumer consumer2 = new Consumer("consumer2", "in", false);
        integralJob.addComponent(consumer2);

        assertThatThrownBy(() -> integralJob.connect(producer, "out", consumer2, "in")).isInstanceOf(IllegalArgumentException.class);
    }

    @Test
    public void testCheckConnections() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);

        integralJob.connect(producer, "out", consumer, "in");
        integralJob.checkConnections();
    }

    @Test
    public void testCheckConnections_missingOutConnection() {
        integralJob.addComponent(producer);
        assertThatThrownBy(() -> integralJob.checkConnections()).isInstanceOf(IllegalStateException.class);
    }

    @Test
    public void testCheckConnections_missingInConnection() {
        integralJob.addComponent(consumer);
        assertThatThrownBy(() -> integralJob.checkConnections()).isInstanceOf(IllegalStateException.class);
    }

    @Test
    public void addMonitorByOutput() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);
        integralJob.connect(producer, "out", consumer, "in");
        integralJob.addMonitorByOutput(producer, "out", (s, d) -> System.out.println(d.getIntegerValue("value")));
    }

    @Test
    public void addMonitorByOutput_OutputNotFound() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);
        integralJob.connect(producer, "out", consumer, "in");
        assertThatThrownBy(
            () -> integralJob.addMonitorByOutput(producer, "out2", (s, d) -> System.out.println(d.getIntegerValue("value")))
        ).isInstanceOf(IllegalArgumentException.class);
    }

    @Test
    public void addMonitorByOutput_OutputNotConnected() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);
        assertThatThrownBy(
            () -> integralJob.addMonitorByOutput(producer, "out", (s, d) -> System.out.println(d.getIntegerValue("value")))
        ).isInstanceOf(IllegalStateException.class);
    }

    @Test
    public void addMonitorByInput() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);
        integralJob.connect(producer, "out", consumer, "in");
        integralJob.addMonitorByInput(consumer, "in", (s, d) -> System.out.println(d.getIntegerValue("value")));
    }

    @Test
    public void addMonitorByInput_InputNotFound() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);
        integralJob.connect(producer, "out", consumer, "in");
        assertThatThrownBy(
            () -> integralJob.addMonitorByInput(consumer, "in2", (s, d) -> System.out.println(d.getIntegerValue("value")))
        ).isInstanceOf(IllegalArgumentException.class);
    }

    @Test
    public void addMonitorByInput_InputNotConnected() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);
        assertThatThrownBy(
            () -> integralJob.addMonitorByInput(consumer, "in", (s, d) -> System.out.println(d.getIntegerValue("value")))
        ).isInstanceOf(IllegalStateException.class);
    }

    @Test
    public void testRun() {
        integralJob.addComponent(producer);
        integralJob.addComponent(consumer);

        integralJob.connect(producer, "out", consumer, "in");
        integralJob.run();

        assertThat(producer.count).isEqualTo(11);
        assertThat(consumer.count).isEqualTo(45);
    }

    @Test
    public void testRun_DataStuck() {
        final Sequence sequence = new Sequence("seq");
        final DeadEnd deadEnd = new DeadEnd("deadEnd");

        integralJob.addComponent(sequence);
        integralJob.addComponent(deadEnd);
        integralJob.connect(sequence, "out", deadEnd, "in");

        assertThatThrownBy(() -> integralJob.run()).isInstanceOf(IllegalStateException.class);
    }

    private final class Producer extends ProducerComponent {

        final String outputName;
        int count = 0;

        public Producer(String name, String outputName) {
            super(name);
            this.outputName = outputName;
            addOutput(outputName, new Schema(new SchemaColumn("value", SIMPLE_NULLABLE_INTEGER_TYPE)));
        }

        @Override
        protected void runInternal() {
            if (count < 10) {
                addToOutput(outputName, new DataRow(java.util.Collections.singletonMap("value", new DataCell((long) count))));
            } else {
                finishProcessing();
            }
            count++;
        }

        @Override
        public void onJobStarted() {
        }
    }

    private final class Consumer extends ConsumerComponent {

        final String inputName;
        int count = 0;

        public Consumer(String name, String inputName, boolean allowNull) {
            super(name);

            this.inputName = inputName;
            addInput(inputName, new Schema(new SchemaColumn("value", new IntegerType(allowNull, null, null))));

        }

        @Override
        protected void runInternal() {
            final DataRow dataRow = pollInputData(inputName);
            if (dataRow != DataRow.EOI) {
                count += dataRow.getIntegerValue("value");
            }
        }

        @Override
        public void onJobStarted() {
        }
    }

    private final class Sequence extends ProducerComponent {

        long count = 0;

        public Sequence(String name) {
            super(name);

            addOutput("out", new Schema(new SchemaColumn("value", SIMPLE_INTEGER_TYPE)));
        }

        @Override
        protected void runInternal() throws ComponentException {
            addToOutput("out", new DataRow(java.util.Collections.singletonMap("value", new DataCell(count++))));
        }
    }

    private final class DeadEnd extends ConsumerComponent {
        public DeadEnd(String name) {
            super(name);

            addInput("in", new Schema(new SchemaColumn("value", SIMPLE_INTEGER_TYPE)));
        }

        @Override
        protected void runInternal() throws ComponentException {
        }
    }
}