package de.arxsilex.maptools.converter.core.pbf

import java.util.concurrent.atomic.AtomicInteger
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue

class BoundedOrderedParallelPipelineTest {
    @Test
    fun `processes concurrently but emits in source order`() {
        val active = AtomicInteger()
        val maximumActive = AtomicInteger()
        val emitted = mutableListOf<Int>()

        BoundedOrderedParallelPipeline.process(
            inputs = (0..11).toList(),
            options = ParallelPbfDecodeOptions(workerCount = 4, maximumBufferedBlocks = 6),
            isCancelled = { false },
            transform = { value ->
                val nowActive = active.incrementAndGet()
                maximumActive.accumulateAndGet(nowActive, ::maxOf)
                try {
                    Thread.sleep(((12 - value) % 4 * 8L) + 5L)
                    value
                } finally {
                    active.decrementAndGet()
                }
            },
            onOrderedResult = { _, value -> emitted += value },
        )

        assertEquals((0..11).toList(), emitted)
        assertTrue(maximumActive.get() > 1)
        assertTrue(maximumActive.get() <= 4)
    }
}
