diff options
Diffstat (limited to 'sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/arrow/effects.kt')
-rw-r--r-- | sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/arrow/effects.kt | 69 |
1 files changed, 69 insertions, 0 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 new file mode 100644 index 00000000..05d13094 --- /dev/null +++ b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/arrow/effects.kt @@ -0,0 +1,69 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2018 NOKIA + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ +package org.onap.dcae.collectors.veshv.utils.arrow + +import arrow.core.Either +import arrow.core.Left +import arrow.core.Right +import arrow.effects.IO +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import kotlin.system.exitProcess + +/** + * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> + * @since June 2018 + */ + +sealed class ExitCode { + abstract val code: Int + + fun io() = IO { + exitProcess(code) + } +} + +object ExitSuccess : ExitCode() { + override val code = 0 +} + +data class ExitFailure(override val code: Int) : ExitCode() + +fun Either<IO<Unit>, IO<Unit>>.unsafeRunEitherSync(onError: (Throwable) -> ExitCode, onSuccess: () -> Unit) = + flatten().attempt().unsafeRunSync().fold({ onError(it).io().unsafeRunSync() }, { onSuccess() }) + + +fun IO<Any>.unit() = map { Unit } + +fun <T> Mono<T>.asIo() = IO.async<T> { callback -> + subscribe({ + callback(Right(it)) + }, { + callback(Left(it)) + }) +} + +fun <T> Flux<IO<T>>.evaluateIo(): Flux<T> = + flatMap { io -> + io.attempt().unsafeRunSync().fold( + { Flux.error<T>(it) }, + { Flux.just<T>(it) } + ) + } |