@FlowPreview
@ExperimentalCoroutinesApi
class FeedbackLoopUnitTest {

    private sealed class MockState : State {
        data class StateA(val round: Int) : MockState()
        data class StateB(val round: Int) : MockState()
        data class StateC(val round: Int) : MockState()
        data class StateD(val round: Int) : MockState()
    }

    private sealed class MockEvent : Event {
        object Next : MockEvent()
    }

    private val testDispatcher = TestCoroutineDispatcher()

    @AfterEach
    fun resetTestDispatcher() {
        testDispatcher.cleanupTestCoroutines()
    }

    @ObsoleteCoroutinesApi
    @RepeatedTest(value=200)
    @DisplayName("Assemble makes a stream that respects the schedulers used in each feedback and in the reducer")
    fun assemble_make_a_stream_that_respect_schedulers() = testDispatcher.runBlockingTest {
        val expectedFeedbackAThread = "FEEDBACKA-${UUID.randomUUID()}"
        val expectedFeedbackBThread = "FEEDBACKB-${UUID.randomUUID()}"
        var expectedReducerThreadId = -1L

        val receivedFeedbackAThreads = mutableListOf<String>()
        val receivedFeedbackBThreads = mutableListOf<String>()
        val receivedReducerThreads = mutableListOf<Long>()

        val sideEffectDispatcherA = newSingleThreadContext(expectedFeedbackAThread)
        val sideEffectDispatcherB = newSingleThreadContext(expectedFeedbackBThread)
        val loopDispatcher = TestCoroutineDispatcher()

        // Getting the Thread id for the loopDispatcher, so we can compare it with the
        // one retrieved in the reducer
        loopDispatcher.invoke {
            expectedReducerThreadId = Thread.currentThread().id
        }

        val feedbackA: (Flow<MockState>) -> Flow<MockEvent> = {
            it.flatMapConcat {
                flowOf("")
                    .onEach { receivedFeedbackAThreads.add(Thread.currentThread().name) }
                    .map { MockEvent.Next }
                    .flowOn(sideEffectDispatcherA)
            }
        }

        val feedbackB: (Flow<MockState>) -> Flow<MockEvent> = {
            it.flatMapConcat {
                flowOf("")
                    .onEach { receivedFeedbackBThreads.add(Thread.currentThread().name) }
                    .map { MockEvent.Next }
                    .flowOn(sideEffectDispatcherB)
            }
        }

        // Given: a reducer that records its execution queue
        val reducer: suspend (MockState, MockEvent) -> MockState = { state, _ ->
            receivedReducerThreads.add(Thread.currentThread().id)
            when (state) {
                is MockState.StateA -> MockState.StateB(state.round + 1)
                is MockState.StateB -> MockState.StateA(state.round + 1)
                else -> state
            }
        }

        // here is the feedback loop definition
        val stateSharedFlow = MutableSharedFlow<MockState>(1)

        val sut = merge(feedbackA(stateSharedFlow), feedbackB(stateSharedFlow))
            .catch {
                Timber.e(it)
                emptyFlow<MockEvent>()
            }
            .scan(MockState.StateA(1), reducer)
            .onEach { stateSharedFlow.emit(it) }
            .flowOn(loopDispatcher)

        // When: executing the loop on a specific scheduler
        val job = launch(loopDispatcher) {
            sut.take(3).collect()
        }

        job.cancel()

        // Then: the feedbacks and reducer happen on the expected schedulers
        Truth.assertThat(receivedFeedbackAThreads).isNotEmpty()
        Truth.assertThat(receivedFeedbackBThreads).isNotEmpty()
        Truth.assertThat(receivedReducerThreads).isNotEmpty()

        receivedFeedbackAThreads.forEach { Truth.assertThat(it).startsWith(expectedFeedbackAThread) }
        receivedFeedbackBThreads.forEach { Truth.assertThat(it).startsWith(expectedFeedbackBThread) }
        receivedReducerThreads.forEach { Truth.assertThat(it).isEqualTo(expectedReducerThreadId) }
    }
}