/
AwaitCancellationStressTest.kt
43 lines (37 loc) · 1.14 KB
/
AwaitCancellationStressTest.kt
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
/*
* Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
*/
package kotlinx.coroutines.reactive
import kotlinx.coroutines.*
import org.junit.*
import org.reactivestreams.*
import java.util.concurrent.locks.*
/**
* This test checks implementation of rule 2.7 for await methods - serial execution of subscription methods
*/
class AwaitCancellationStressTest : TestBase() {
private val iterations = 10_000 * stressTestMultiplier
@Test
fun testAwaitCancellationOrder() = runTest {
repeat(iterations) {
val job = launch(Dispatchers.Default) {
testPublisher().awaitFirst()
}
job.cancelAndJoin()
}
}
private fun testPublisher() = Publisher<Int> { s ->
val lock = ReentrantLock()
s.onSubscribe(object : Subscription {
override fun request(n: Long) {
check(lock.tryLock())
s.onNext(42)
lock.unlock()
}
override fun cancel() {
check(lock.tryLock())
lock.unlock()
}
})
}
}