blob: 1430dbf3812749a7e1156b80f987a2e896454cfa (
plain)
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
|
/*
* Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
*/
package kotlinx.coroutines.rx2
import io.reactivex.*
import io.reactivex.exceptions.*
import io.reactivex.plugins.*
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.reactive.*
import org.junit.Test
import java.io.*
import kotlin.test.*
// Check that exception is not leaked to the global exception handler
class LeakedExceptionTest : TestBase() {
private val handler: (Throwable) -> Unit =
{ assertTrue { it is UndeliverableException && it.cause is TestException } }
@Test
fun testSingle() = withExceptionHandler(handler) {
val flow = rxSingle<Unit> { throw TestException() }.toFlowable().asFlow()
runBlocking {
repeat(10000) {
combine(flow, flow) { _, _ -> Unit }
.catch {}
.collect { }
}
}
}
@Test
fun testObservable() = withExceptionHandler(handler) {
val flow = rxObservable<Unit> { throw TestException() }.toFlowable(BackpressureStrategy.BUFFER).asFlow()
runBlocking {
repeat(10000) {
combine(flow, flow) { _, _ -> Unit }
.catch {}
.collect { }
}
}
}
@Test
fun testFlowable() = withExceptionHandler(handler) {
val flow = rxFlowable<Unit> { throw TestException() }.asFlow()
runBlocking {
repeat(10000) {
combine(flow, flow) { _, _ -> Unit }
.catch {}
.collect { }
}
}
}
}
|
