/
MonoK.kt
70 lines (55 loc) · 2.06 KB
/
MonoK.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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
package arrow.effects
import arrow.core.Either
import arrow.core.Either.Left
import arrow.core.Either.Right
import arrow.effects.CoroutineContextReactorScheduler.asScheduler
import arrow.effects.typeclasses.Proc
import arrow.higherkind
import reactor.core.publisher.Mono
import reactor.core.publisher.MonoSink
import kotlin.coroutines.experimental.CoroutineContext
fun <A> Mono<A>.k(): MonoK<A> = MonoK(this)
fun <A> MonoKOf<A>.value(): Mono<A> =
this.fix().mono
@higherkind
data class MonoK<A>(val mono: Mono<A>) : MonoKOf<A>, MonoKKindedJ<A> {
fun <B> map(f: (A) -> B): MonoK<B> =
mono.map(f).k()
fun <B> ap(fa: MonoKOf<(A) -> B>): MonoK<B> =
flatMap { a -> fa.fix().map { ff -> ff(a) } }
fun <B> flatMap(f: (A) -> MonoKOf<B>): MonoK<B> =
mono.flatMap { f(it).fix().mono }.k()
fun handleErrorWith(function: (Throwable) -> MonoK<A>): MonoK<A> =
mono.onErrorResume { t: Throwable -> function(t).mono }.k()
fun continueOn(ctx: CoroutineContext): MonoK<A> =
mono.publishOn(ctx.asScheduler()).k()
fun runAsync(cb: (Either<Throwable, A>) -> MonoKOf<Unit>): MonoK<Unit> =
mono.flatMap { cb(Right(it)).value() }.onErrorResume { cb(Left(it)).value() }.k()
companion object {
fun <A> just(a: A): MonoK<A> =
Mono.just(a).k()
fun <A> raiseError(t: Throwable): MonoK<A> =
Mono.error<A>(t).k()
operator fun <A> invoke(fa: () -> A): MonoK<A> =
defer { just(fa()) }
fun <A> defer(fa: () -> MonoKOf<A>): MonoK<A> =
Mono.defer { fa().value() }.k()
fun <A> async(fa: Proc<A>): MonoK<A> =
Mono.create({ emitter: MonoSink<A> ->
fa { either: Either<Throwable, A> ->
either.fold({
emitter.error(it)
}, {
emitter.success(it)
})
}
}).k()
tailrec fun <A, B> tailRecM(a: A, f: (A) -> MonoKOf<Either<A, B>>): MonoK<B> {
val either = f(a).fix().value().block()
return when (either) {
is Either.Left -> tailRecM(either.a, f)
is Either.Right -> Mono.just(either.b).k()
}
}
}
}