diff options
Diffstat (limited to 'sources/hv-collector-utils')
| -rw-r--r-- | sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/arrow/effects.kt | 8 |
1 files changed, 5 insertions, 3 deletions
diff --git a/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/arrow/effects.kt b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/arrow/effects.kt index 290ef72c..56825221 100644 --- a/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/arrow/effects.kt +++ b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/arrow/effects.kt @@ -23,11 +23,10 @@ import arrow.core.Either import arrow.core.Left import arrow.core.Right import arrow.effects.IO -import arrow.effects.fix -import arrow.effects.instances.io.monadError.monadError -import arrow.typeclasses.binding +import org.reactivestreams.Publisher import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.core.publisher.toMono import kotlin.system.exitProcess /** @@ -62,6 +61,9 @@ fun <T> Mono<T>.asIo() = IO.async<T> { callback -> }) } +fun <T> Publisher<T>.then(callback: () -> Unit): Mono<Unit> = + toMono().then(Mono.fromCallable(callback)) + fun <T> Flux<IO<T>>.evaluateIo(): Flux<T> = flatMap { io -> io.attempt().unsafeRunSync().fold( |