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
|
/*
* Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
*/
package kotlinx.coroutines.reactor
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
fun <T> checkMonoValue(
mono: Mono<T>,
checker: (T) -> Unit
) {
val monoValue = mono.block()
checker(monoValue)
}
fun checkErroneous(
mono: Mono<*>,
checker: (Throwable) -> Unit
) {
try {
mono.block()
error("Should have failed")
} catch (e: Throwable) {
checker(e)
}
}
fun <T> checkSingleValue(
flux: Flux<T>,
checker: (T) -> Unit
) {
val singleValue = flux.toIterable().single()
checker(singleValue)
}
fun checkErroneous(
flux: Flux<*>,
checker: (Throwable) -> Unit
) {
val singleNotification = flux.materialize().toIterable().single()
checker(singleNotification.throwable)
}
|
