Coroutines newbie question: I want to test the fol...

# coroutines

[David Kubecka]

Coroutines newbie question: I want to test the following code:

@Service
class AsyncExecutionManager(val taskExecutor: AsyncTaskExecutor) {

    private val jobs = ConcurrentHashMap<String, JobStatus>()

    fun <T> startJob(jobId: String, job: () -> T) {
        val previousStatus = jobs.put(jobId, JobStatus.RUNNING)
        if (previousStatus != JobStatus.RUNNING) {
            taskExecutor.submit {
                job()
            }
        }
    }

Namely, I want to check that the same job cannot be run twice at the same time. To do that, I mock AsyncTaskExecutor#submit (from Spring) by starting a thread that just runs the given job lambda, catches exceptions, and stores them to a list which is then checked in the test. The mock looks as follows:

private val asyncManager = AsyncExecutionManager(
        mockk {
            val slot = slot<Runnable>()
            every { submit(capture(slot)) } answers {
                thread {
                    try {
                        slot.captured.run()
                    } catch (e: Exception) {
                        errors.add(e)
                    }
                }
                CompletableFuture.completedFuture(Unit)
            }
        },
    )

And the test itself:

@Test
    fun `starting the same task concurrently`() {
//        asyncManager.startJob(job1) { error("should not happen") }
        asyncManager.startJob(job1) { Thread.sleep(100) }
        asyncManager.startJob(job1) { error("should not happen") }
        expectThat(asyncManager.getJobStatus(job1)).isEqualTo(JobStatus.RUNNING)
        Thread.sleep(100)
        expectThat(errors).isEmpty()
    }

The test passes on my machine but I'm not happy about two of its aspects:
• The last sleep which must be there to "wait" for the error-throwing threads to catch and handle the exceptions. Without the sleep, if I uncomment the first thread, the test passes even if it shouldn't.
• Using threads instead of coroutines. I wonder if there's a way how to rewrite my test to start a coroutine instead of a thread in the mocked AsyncTaskExecutor#submit . The problem is that the method doesn't provide a coroutine scope and I don't know how to provide one. I tried making job suspending and using runBlocking inside the real submit lambda but then the jobs in my test don't run concurrently anymore.
So basically there are two questions:
• Can coroutines be used in my setup of mocking AsyncTaskExecutor#submit ?
• Is there a more robust way to test my code's correctness, other than collecting errors and waiting "sufficient" time? This question is probably not Kotlin-specific but I hope it might be related to the first question, or even there might be special Kotlin tooling for these kinds of situations.

[ephemient]

how about adapting the coroutine dispatcher to an executor? no mocks required

@Test
fun test() = runTest {
    val dispatcher = coroutineContext[CoroutineDispatcher]!!
    val taskExecutor = ConcurrentTaskExecutor(dispatcher.asExecutor())
    val asyncManager = AsyncExecutionManager(taskExecutor)

[David Kubecka]

Yes, that allows me to replace the mock with explicit taskExecutor! Still, I cannot use coroutine elements inside my jobs (e.g. delay instead of sleep). I guess there's no way that can be achieved while using AsyncTaskExecutor#submit , right?

[David Kubecka]

Moreover, it doesn't solve my other problem of detecting a job error in a more robust way than sleeping. In fact, the coroutine solution seems to be a bit further away from exactly because there are no mocks now. Any ideas about that?

[ephemient]

right, there no (sane) way use coroutines inside those jobs. but it should give you a way to avoid sleeping: runCurrent() will advance all coroutines to the current time. actually the test dispatcher is single threaded, for consistent behavior across runs, so your jobs should be run anyway

[David Kubecka]

But I don't want single-threaded dispatch. Quite the contrary, I want to allow concurrent execution of jobs. Just this particular test case (I have also others) tests that the same job shouldn't run concurrently multiple times. That's why I have the sleep there - to simulate long running task.

[ephemient]

hmm. you could use multi-threaded dispatchers instead of runTest's TestDispatcher, but honestly you should care more about testing concurrency than parallelism

[David Kubecka]

What do you mean by that? Do you suggest I should structure my test differently?

[ephemient]

yes. even without using coroutines, you can use various java.util.concurrent utilities such as CyclicBarrier or Phaser to observe and control what each runnable is doing, instead of relying on thread sleeps

[David Kubecka]

You are probably talking about the first sleep and I know I can replace that with various synchronization primitives. My original question, though, was about the second sleep which is needed simply to wait for both the second thread (with error) and its handler in the mock to finish so that I can check whether there were any errors.