aboutsummaryrefslogtreecommitdiffstats
path: root/sources/hv-collector-utils
diff options
context:
space:
mode:
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.kt8
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(