package de.accso.flexinale.common.api.eventbus;

import de.accso.flexinale.common.api.eventbus.PublisherAndSubscriberTestdata.*;
import de.accso.flexinale.common.api.event.EventTestdata.MyNumberEvent;
import de.accso.flexinale.common.infrastructure.eventbus.InMemorySyncEventBusFactory;
import de.accso.flexinale.common.infrastructure.eventbus.InMemorySyncEventBusSpy;
import de.accso.flexinale.common.infrastructure.eventbus.InMemorySyncEventBusSpyFactory;
import de.accso.flexinale.common.shared_kernel.DeveloperMistakeException;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;

import java.util.List;
import java.util.Set;

import static de.accso.flexinale.common.api.eventbus.EventSubscriptionAtStart.*;
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;

class InMemorySyncEventBusAndSpySubscriptionTest {

    @Test
    void testSubscriptionAtStart_START_READING_FROM_NOW() {
        // arrange
        int numEventsToPublish = 100;

        InMemorySyncEventBusFactory factory = new InMemorySyncEventBusFactory();
        EventBus<MyNumberEvent> sut = factory.createOrGetEventBusFor(MyNumberEvent.class);

        internal_testSubscriberReceivesMultipleEvents_START_READING_FROM_NOW(numEventsToPublish, sut);
    }

    @Test
    void testSubscriptionAtStart_START_READING_FROM_NOW_Spy() {
        // arrange
        int numEventsToPublish = 100;

        InMemorySyncEventBusFactory factory = new InMemorySyncEventBusSpyFactory();
        EventBus<MyNumberEvent> sut = factory.createOrGetEventBusFor(MyNumberEvent.class);

        MyNumberEvent myEvent = internal_testSubscriberReceivesMultipleEvents_START_READING_FROM_NOW(numEventsToPublish, sut);

        // ... and assert spy functionality
        InMemorySyncEventBusSpy<MyNumberEvent> spy = (InMemorySyncEventBusSpy<MyNumberEvent>) sut;
        List<MyNumberEvent> allEvents = spy.allEvents();
        assertThat(allEvents.getLast()).isEqualTo(myEvent);
        assertThat(spy.size()).isEqualTo(numEventsToPublish);
    }

    private static MyNumberEvent internal_testSubscriberReceivesMultipleEvents_START_READING_FROM_NOW(
            final int numEventsToPublish, final EventBus<MyNumberEvent> sut) {

        MyNumberEventPublisher myPublisher = new MyNumberEventPublisher(sut);

        MyNumberEventSubscriber mySubscriberSubscribedBeforePublish = new MyNumberEventSubscriber(true);
        MyNumberEventSubscriber mySubscriberSubscribedAfterPublish  = new MyNumberEventSubscriber(true);

        // act
        sut.subscribe(MyNumberEvent.class, mySubscriberSubscribedBeforePublish, START_READING_FROM_NOW);
        MyNumberEvent myEvent = null;
        for (int counter = 0; counter < numEventsToPublish; counter++) {
            myEvent = new MyNumberEvent(counter, counter);
            myPublisher.post(MyNumberEvent.class, myEvent);
        }
        sut.subscribe(MyNumberEvent.class, mySubscriberSubscribedAfterPublish, START_READING_FROM_NOW);

        // assert
        assertThat(mySubscriberSubscribedBeforePublish.lastReceivedEvent).isEqualTo(myEvent);
        assertThat(mySubscriberSubscribedBeforePublish.counterOfRetrievedEvents).isEqualTo(numEventsToPublish);

        assertThat(mySubscriberSubscribedAfterPublish.lastReceivedEvent).isNull();
        assertThat(mySubscriberSubscribedAfterPublish.counterOfRetrievedEvents).isEqualTo(0);

        return myEvent;
    }

    // -----------------------------------------------------------------------------------------------------------

    @Test
    void testSubscriptionAtStart_START_READING_FROM_BEGINNING_throwsException() {
        // arrange
        InMemorySyncEventBusFactory factory = new InMemorySyncEventBusFactory();
        EventBus<MyNumberEvent> sut = factory.createOrGetEventBusFor(MyNumberEvent.class);

        MyNumberEventSubscriber mySubscriber = new MyNumberEventSubscriber(true);

        // act and assert
        Assertions.assertThatThrownBy(() -> {
            sut.subscribe(MyNumberEvent.class, mySubscriber, START_READING_FROM_BEGINNING);
        })
        .isInstanceOf(DeveloperMistakeException.class);
    }

    @Test
    void testSubscriptionAtStart_START_READING_FROM_BEGINNING_Spy() {
        // arrange
        int numEventsToPublish = 100;

        InMemorySyncEventBusFactory factory = new InMemorySyncEventBusSpyFactory();
        EventBus<MyNumberEvent> sut = factory.createOrGetEventBusFor(MyNumberEvent.class);
        InMemorySyncEventBusSpy<MyNumberEvent> spy = (InMemorySyncEventBusSpy<MyNumberEvent>) sut;
        MyNumberEventPublisher myPublisher = new MyNumberEventPublisher(sut);

        MyNumberEventSubscriber mySubscriberSubscribedBeforePublish = new MyNumberEventSubscriber(true);
        MyNumberEventSubscriber mySubscriberSubscribedAfterPublish  = new MyNumberEventSubscriber(true);

        // act - one subscriber before the publish, the other afterwards - with START_READING_FROM_BEGINNING both get the same
        sut.subscribe(MyNumberEvent.class, mySubscriberSubscribedBeforePublish, START_READING_FROM_BEGINNING);
        MyNumberEvent myEvent = null;
        for (int counter = 0; counter < numEventsToPublish; counter++) {
            myEvent = new MyNumberEvent(counter, counter);
            myPublisher.post(MyNumberEvent.class, myEvent);
        }
        sut.subscribe(MyNumberEvent.class, mySubscriberSubscribedAfterPublish, START_READING_FROM_BEGINNING);

        // assert
        assertThat(mySubscriberSubscribedBeforePublish.lastReceivedEvent).isEqualTo(myEvent);
        assertThat(mySubscriberSubscribedBeforePublish.counterOfRetrievedEvents).isEqualTo(numEventsToPublish);

        assertThat(mySubscriberSubscribedAfterPublish.lastReceivedEvent).isEqualTo(myEvent);
        assertThat(mySubscriberSubscribedAfterPublish.counterOfRetrievedEvents).isEqualTo(numEventsToPublish);

        // ... and assert spy functionality
        List<MyNumberEvent> allEvents = spy.allEvents();
        assertThat(allEvents.getLast()).isEqualTo(myEvent);
        assertThat(spy.size()).isEqualTo(numEventsToPublish);
    }

    // -----------------------------------------------------------------------------------------------------------

    @Test
    void testSubscriptionAtStart_START_READING_FROM_LAST_TIME_throwsException() {
        // arrange
        InMemorySyncEventBusFactory factory = new InMemorySyncEventBusFactory();
        EventBus<MyNumberEvent> sut = factory.createOrGetEventBusFor(MyNumberEvent.class);

        MyNumberEventSubscriber mySubscriber = new MyNumberEventSubscriber(true);

        // act and assert
        Assertions.assertThatThrownBy(() -> {
            sut.subscribe(MyNumberEvent.class, mySubscriber, START_READING_FROM_LAST_TIME);
        })
        .isInstanceOf(DeveloperMistakeException.class);
    }

    @Test
    void testSubscriptionAtStart_START_READING_FROM_LAST_TIME_Spy_oneConsumer() {
        // arrange
        int numEventsToPublish = 100;
        int counter;

        InMemorySyncEventBusFactory factory = new InMemorySyncEventBusSpyFactory();
        EventBus<MyNumberEvent> sut = factory.createOrGetEventBusFor(MyNumberEvent.class);
        InMemorySyncEventBusSpy<MyNumberEvent> spy = (InMemorySyncEventBusSpy<MyNumberEvent>) sut;
        MyNumberEventPublisher myPublisher = new MyNumberEventPublisher(sut);

        MyNumberEventSubscriber mySubscriber = new MyNumberEventSubscriber(true);

        // act

        // 1) subscribe
        sut.subscribe(MyNumberEvent.class, mySubscriber, START_READING_FROM_LAST_TIME);

        // 2) sent first half
        MyNumberEvent myEvent = null;
        for (counter = 0; counter < numEventsToPublish / 2; counter++) {
            myEvent = new MyNumberEvent(counter, counter);
            myPublisher.post(MyNumberEvent.class, myEvent);
        }

        assertThat(mySubscriber.lastReceivedEvent).isEqualTo(myEvent);
        assertThat(mySubscriber.counterOfRetrievedEvents).isEqualTo(numEventsToPublish / 2);

        // 3) unsubscribe
        sut.unsubscribe(MyNumberEvent.class, mySubscriber);

        // 4) sent second half
        for (; counter < numEventsToPublish; counter++) {
            myEvent = new MyNumberEvent(counter, counter);
            myPublisher.post(MyNumberEvent.class, myEvent);
        }

        // 5) subscribe again ->  get the second half
        sut.subscribe(MyNumberEvent.class, mySubscriber, START_READING_FROM_LAST_TIME);

        // 6) unsubscribe
        sut.unsubscribe(MyNumberEvent.class, mySubscriber);
        assertThat(mySubscriber.lastReceivedEvent).isEqualTo(myEvent);
        assertThat(mySubscriber.counterOfRetrievedEvents).isEqualTo(numEventsToPublish);

        // 7) subscribe again ->  get nothing else
        sut.subscribe(MyNumberEvent.class, mySubscriber, START_READING_FROM_LAST_TIME);
        assertThat(mySubscriber.lastReceivedEvent).isEqualTo(myEvent);
        assertThat(mySubscriber.counterOfRetrievedEvents).isEqualTo(numEventsToPublish);

        // ... and assert spy functionality
        List<MyNumberEvent> allEvents = spy.allEvents();
        assertThat(allEvents.getLast()).isEqualTo(myEvent);
        assertThat(spy.size()).isEqualTo(numEventsToPublish);
    }

    @Test
    void testSubscriptionAtStart_START_READING_FROM_LAST_TIME_Spy_twoConsumersInOneGroup() {
        // arrange
        int numEventsToPublish = 100;
        int counter;

        InMemorySyncEventBusFactory factory = new InMemorySyncEventBusSpyFactory();
        EventBus<MyNumberEvent> sut = factory.createOrGetEventBusFor(MyNumberEvent.class);
        InMemorySyncEventBusSpy<MyNumberEvent> spy = (InMemorySyncEventBusSpy<MyNumberEvent>) sut;

        MyNumberEventPublisher myPublisher = new MyNumberEventPublisher(sut);

        MyNumberEventSubscriber mySubscriber1 = new MyNumberEventSubscriber(true);
        MyNumberEventSubscriber mySubscriber2 = new MyNumberEventSubscriber(true);

        // act

        // 1) subscribe
        sut.subscribe(MyNumberEvent.class, mySubscriber1, START_READING_FROM_LAST_TIME);
        sut.subscribe(MyNumberEvent.class, mySubscriber2, START_READING_FROM_LAST_TIME);

        // 2) sent first half
        MyNumberEvent myEvent50 = null;
        for (counter = 0; counter < numEventsToPublish / 2; counter++) {
            myEvent50 = new MyNumberEvent(counter, counter);
            myPublisher.post(MyNumberEvent.class, myEvent50);
        }

        // only one of the subscribers has the last event
        Set<MyNumberEvent> lastReceivedEvents = Set.of(mySubscriber1.lastReceivedEvent, mySubscriber2.lastReceivedEvent);
        assertThat(lastReceivedEvents.size()).isEqualTo(2); // Set.of() already guarantees that the two elements are different (otherwise "java.lang.IllegalArgumentException: duplicate element")
        assertThat(lastReceivedEvents.contains(myEvent50)).isTrue();

        // both subscribers received all events in total
        assertThat(mySubscriber1.counterOfRetrievedEvents + mySubscriber2.counterOfRetrievedEvents)
                .isEqualTo(numEventsToPublish / 2);

        int counterOfRetrievedEventsOfSubscriber1AfterFirstHalf    = mySubscriber1.counterOfRetrievedEvents;
        MyNumberEvent mySubscriber1LastReceivedEventAfterFirstHalf = mySubscriber1.lastReceivedEvent;
        int counterOfRetrievedEventsOfSubscriber2AfterFirstHalf    = mySubscriber2.counterOfRetrievedEvents;
        MyNumberEvent mySubscriber2LastReceivedEventAfterFirstHalf = mySubscriber2.lastReceivedEvent;

        // 3) unsubscribe
        sut.unsubscribe(MyNumberEvent.class, mySubscriber1);
        sut.unsubscribe(MyNumberEvent.class, mySubscriber2);

        // 4) sent second half
        MyNumberEvent myEvent100 = null;
        for (; counter < numEventsToPublish; counter++) {
            myEvent100 = new MyNumberEvent(counter, counter);
            myPublisher.post(MyNumberEvent.class, myEvent100);
        }

        // 5) subscribe subscriber 1 again ->  gets the second half completely
        sut.subscribe(MyNumberEvent.class, mySubscriber1, START_READING_FROM_LAST_TIME);
        assertThat(mySubscriber1.lastReceivedEvent).isEqualTo(myEvent100);
        assertThat(mySubscriber1.counterOfRetrievedEvents)
                .isEqualTo(counterOfRetrievedEventsOfSubscriber1AfterFirstHalf + numEventsToPublish/2);

        // the second subscriber get's nothing as both subscribers are in one group
        // (wait fail, when the following subscribe comes to early -> then add to wait a little here)
        sut.subscribe(MyNumberEvent.class, mySubscriber2, START_READING_FROM_LAST_TIME);
        assertThat(mySubscriber2.lastReceivedEvent).isEqualTo(mySubscriber2LastReceivedEventAfterFirstHalf);
        assertThat(mySubscriber2.counterOfRetrievedEvents)
                .isEqualTo(counterOfRetrievedEventsOfSubscriber2AfterFirstHalf);

        int counterOfRetrievedEventsOfSubscriber1AfterSecondHalf = mySubscriber1.counterOfRetrievedEvents;
        int counterOfRetrievedEventsOfSubscriber2AfterSecondHalf = mySubscriber2.counterOfRetrievedEvents;

        assertThat(counterOfRetrievedEventsOfSubscriber1AfterSecondHalf + counterOfRetrievedEventsOfSubscriber2AfterSecondHalf)
                .isEqualTo(numEventsToPublish);

        // 6) unsubscribe
        sut.unsubscribe(MyNumberEvent.class, mySubscriber1);
        sut.unsubscribe(MyNumberEvent.class, mySubscriber2);

        // 7) subscribe again ->  get nothing else
        sut.subscribe(MyNumberEvent.class, mySubscriber1, START_READING_FROM_LAST_TIME);
        assertThat(mySubscriber1.lastReceivedEvent).isEqualTo(myEvent100);
        assertThat(mySubscriber1.counterOfRetrievedEvents)
                .isEqualTo(counterOfRetrievedEventsOfSubscriber1AfterSecondHalf);
        sut.subscribe(MyNumberEvent.class, mySubscriber2, START_READING_FROM_LAST_TIME);
        assertThat(mySubscriber2.lastReceivedEvent)
                .isEqualTo(mySubscriber2LastReceivedEventAfterFirstHalf);
        assertThat(mySubscriber2.counterOfRetrievedEvents)
                .isEqualTo(counterOfRetrievedEventsOfSubscriber2AfterSecondHalf);

        // ... and assert spy functionality
        List<MyNumberEvent> allEvents = spy.allEvents();
        assertThat(allEvents.getLast()).isEqualTo(myEvent100);
        assertThat(spy.size()).isEqualTo(numEventsToPublish);
    }

    @Test
    void testSubscriptionAtStart_START_READING_FROM_LAST_TIME_Spy_twoConsumersInSeparateGroups() {
        // arrange
        int numEventsToPublish = 100;
        int counter;

        InMemorySyncEventBusFactory factory = new InMemorySyncEventBusSpyFactory();
        EventBus<MyNumberEvent> sut = factory.createOrGetEventBusFor(MyNumberEvent.class);
        InMemorySyncEventBusSpy<MyNumberEvent> spy = (InMemorySyncEventBusSpy<MyNumberEvent>) sut;

        MyNumberEventSubscriber mySubscriber1 = new MyNumberEventSubscriber(true);
        MyEventSubscriberEachInstanceInItsOwnGroup mySubscriber2 = new MyEventSubscriberEachInstanceInItsOwnGroup(true);
        MyNumberEventPublisher myPublisher = new MyNumberEventPublisher(sut);

        // act

        // 1) subscribe
        sut.subscribe(MyNumberEvent.class, mySubscriber1, START_READING_FROM_LAST_TIME);
        sut.subscribe(MyNumberEvent.class, mySubscriber2, START_READING_FROM_LAST_TIME);

        // 2) sent first half
        MyNumberEvent myEvent50 = null;
        for (counter = 0; counter < numEventsToPublish / 2; counter++) {
            myEvent50 = new MyNumberEvent(counter, counter);
            myPublisher.post(MyNumberEvent.class, myEvent50);
        }

        assertThat(mySubscriber1.lastReceivedEvent).isEqualTo(myEvent50);
        assertThat(mySubscriber1.counterOfRetrievedEvents).isEqualTo(numEventsToPublish / 2);
        assertThat(mySubscriber2.lastReceivedEvent).isEqualTo(myEvent50);
        assertThat(mySubscriber2.counterOfRetrievedEvents).isEqualTo(numEventsToPublish / 2);

        // 3) unsubscribe
        sut.unsubscribe(MyNumberEvent.class, mySubscriber1);
        sut.unsubscribe(MyNumberEvent.class, mySubscriber2);

        // 4) sent second half
        MyNumberEvent myEvent100 = null;
        for (; counter < numEventsToPublish; counter++) {
            myEvent100 = new MyNumberEvent(counter, counter);
            myPublisher.post(MyNumberEvent.class, myEvent100);
        }

        // 5) subscribe again ->  get the second half
        sut.subscribe(MyNumberEvent.class, mySubscriber1, START_READING_FROM_LAST_TIME);
        assertThat(mySubscriber1.lastReceivedEvent).isEqualTo(myEvent100);
        assertThat(mySubscriber1.counterOfRetrievedEvents).isEqualTo(numEventsToPublish);

        // the second subscriber get's same as both subscribers are in separate groups
        sut.subscribe(MyNumberEvent.class, mySubscriber2, START_READING_FROM_LAST_TIME);
        assertThat(mySubscriber2.lastReceivedEvent).isEqualTo(myEvent100);
        assertThat(mySubscriber2.counterOfRetrievedEvents).isEqualTo(numEventsToPublish);

        // 6) unsubscribe
        sut.unsubscribe(MyNumberEvent.class, mySubscriber1);
        sut.unsubscribe(MyNumberEvent.class, mySubscriber2);

        // 7) subscribe again ->  get nothing else
        sut.subscribe(MyNumberEvent.class, mySubscriber1, START_READING_FROM_LAST_TIME);
        assertThat(mySubscriber1.lastReceivedEvent).isEqualTo(myEvent100);
        assertThat(mySubscriber1.counterOfRetrievedEvents).isEqualTo(numEventsToPublish);
        sut.subscribe(MyNumberEvent.class, mySubscriber2, START_READING_FROM_LAST_TIME);
        assertThat(mySubscriber2.lastReceivedEvent).isEqualTo(myEvent100);
        assertThat(mySubscriber2.counterOfRetrievedEvents).isEqualTo(numEventsToPublish);

        // ... and assert spy functionality
        List<MyNumberEvent> allEvents = spy.allEvents();
        assertThat(allEvents.getLast()).isEqualTo(myEvent100);
        assertThat(spy.size()).isEqualTo(numEventsToPublish);
    }

    @Test
    void testSubscriptionAtStart_START_READING_FROM_BEGINNING_and_LAST_TIME_Spy() {
        // arrange
        int numEventsToPublish = 100;
        int counter;

        InMemorySyncEventBusFactory factory = new InMemorySyncEventBusSpyFactory();
        EventBus<MyNumberEvent> sut = factory.createOrGetEventBusFor(MyNumberEvent.class);
        InMemorySyncEventBusSpy<MyNumberEvent> spy = (InMemorySyncEventBusSpy<MyNumberEvent>) sut;

        // false, as we get some events twice
        MyNumberEventSubscriber mySubscriberSubscribedBeforeAndAfterPublish = new MyNumberEventSubscriber(false);
        MyNumberEventPublisher myPublisher = new MyNumberEventPublisher(sut);

        // act

        // 1) subscribe
        sut.subscribe(MyNumberEvent.class, mySubscriberSubscribedBeforeAndAfterPublish, START_READING_FROM_LAST_TIME);

        // 2) sent first half
        MyNumberEvent myEvent = null;
        for (counter = 0; counter < numEventsToPublish / 2; counter++) {
            myEvent = new MyNumberEvent(counter, counter);
            myPublisher.post(MyNumberEvent.class, myEvent);
        }

        // 3) unsubscribe
        sut.unsubscribe(MyNumberEvent.class, mySubscriberSubscribedBeforeAndAfterPublish);

        // 4) sent second half
        for (; counter < numEventsToPublish; counter++) {
            myEvent = new MyNumberEvent(counter, counter);
            myPublisher.post(MyNumberEvent.class, myEvent);
        }

        // 5) subscribe again ->  get it all, i.e. the first half and the second half
        sut.subscribe(MyNumberEvent.class, mySubscriberSubscribedBeforeAndAfterPublish, START_READING_FROM_BEGINNING);

        // assert
        assertThat(mySubscriberSubscribedBeforeAndAfterPublish.lastReceivedEvent).isEqualTo(myEvent);
        assertThat(mySubscriberSubscribedBeforeAndAfterPublish.counterOfRetrievedEvents)
                .isEqualTo(numEventsToPublish/2 + numEventsToPublish);

        // 6) unsubscribe
        sut.unsubscribe(MyNumberEvent.class, mySubscriberSubscribedBeforeAndAfterPublish);

        // 7) subscribe again ->  get nothing else
        sut.subscribe(MyNumberEvent.class, mySubscriberSubscribedBeforeAndAfterPublish, START_READING_FROM_LAST_TIME);

        // assert
        assertThat(mySubscriberSubscribedBeforeAndAfterPublish.lastReceivedEvent).isEqualTo(myEvent);
        assertThat(mySubscriberSubscribedBeforeAndAfterPublish.counterOfRetrievedEvents)
                .isEqualTo(numEventsToPublish/2 + numEventsToPublish);

        // ... and assert spy functionality
        List<MyNumberEvent> allEvents = spy.allEvents();
        assertThat(allEvents.getLast()).isEqualTo(myEvent);
        assertThat(spy.size()).isEqualTo(numEventsToPublish);
    }
}