<properties>
<kotlin.version>1.3.21</kotlin.version>
- <arrow.version>0.8.0</arrow.version>
+ <arrow.version>0.9.0</arrow.version>
<maven-compiler-plugin.version>3.7.0</maven-compiler-plugin.version>
<build-helper-maven-plugin.version>1.7</build-helper-maven-plugin.version>
<jacoco.version>0.8.2</jacoco.version>
- <detekt.version>1.0.0-RC11</detekt.version>
+ <detekt.version>1.0.0-RC14</detekt.version>
<sdk.version>1.1.4-SNAPSHOT</sdk.version>
<!-- Protocol buffers -->
</dependency>
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib-jdk8</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-reflect</artifactId>
+ <artifactId>kotlin-bom</artifactId>
<version>${kotlin.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
</dependency>
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-compiler-embeddable</artifactId>
<version>${kotlin.version}</version>
</dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-script-runtime</artifactId>
- <version>${kotlin.version}</version>
- <scope>runtime</scope>
- </dependency>
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-script-util</artifactId>
<dependency>
<groupId>org.jetbrains.kotlinx</groupId>
<artifactId>kotlinx-coroutines-core</artifactId>
- <version>1.0.0</version>
+ <version>1.1.1</version>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
- <artifactId>arrow-core</artifactId>
+ <artifactId>arrow-core-data</artifactId>
<version>${arrow.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib-jdk7</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
- <artifactId>arrow-syntax</artifactId>
+ <artifactId>arrow-core-extensions</artifactId>
<version>${arrow.version}</version>
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
- <artifactId>arrow-instances-core</artifactId>
+ <artifactId>arrow-extras-data</artifactId>
<version>${arrow.version}</version>
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
- <artifactId>arrow-instances-data</artifactId>
- <version>${arrow.version}</version>
- </dependency>
- <dependency>
- <groupId>io.arrow-kt</groupId>
- <artifactId>arrow-effects</artifactId>
- <version>${arrow.version}</version>
- </dependency>
- <dependency>
- <groupId>io.arrow-kt</groupId>
- <artifactId>arrow-effects-instances</artifactId>
+ <artifactId>arrow-syntax</artifactId>
<version>${arrow.version}</version>
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
- <artifactId>arrow-effects-reactor</artifactId>
+ <artifactId>arrow-typeclasses</artifactId>
<version>${arrow.version}</version>
</dependency>
<dependency>
<dependency>
<groupId>com.nhaarman.mockitokotlin2</groupId>
<artifactId>mockito-kotlin</artifactId>
- <version>2.0.0</version>
+ <version>2.1.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
</dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-reflect</artifactId>
- </dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-test</artifactId>
- <scope>test</scope>
- </dependency>
<dependency>
<groupId>org.jetbrains.spek</groupId>
<artifactId>spek-api</artifactId>
Try { parseArgumentsArray(args) }
.toEither()
.mapLeft { WrongArgumentError(it, cmdLineOptionsList) }
- .map(this::getConfiguration)
+ .map(::getConfiguration)
.flatMap {
it.toEither {
WrongArgumentError(
*/
package org.onap.dcae.collectors.veshv.commandline
-import arrow.core.Option
import org.apache.commons.cli.HelpFormatter
import org.apache.commons.cli.Options
private fun getOptions() = cmdLineOptionsList.map { it.option }.fold(Options(), Options::addOption)
companion object {
- fun generateRequiredParametersNote(cmdLineOptionsList: List<CommandLineOption>): String {
- val requiredParams = Option.fromNullable(cmdLineOptionsList.filter { it.required }
- .takeUnless { it.isEmpty() })
- return requiredParams.fold(
- { "" },
- {
- it.map { commandLineOption -> commandLineOption.option.opt }
+ fun generateRequiredParametersNote(cmdLineOptionsList: List<CommandLineOption>): String =
+ cmdLineOptionsList.filter { it.required }.let { requiredParams ->
+ if (requiredParams.isEmpty())
+ ""
+ else
+ requiredParams.map { commandLineOption -> commandLineOption.option.opt }
.joinToString(prefix = "Required parameters: ", separator = ", ")
- }
- )
- }
+ }
+
}
}
import arrow.core.Option
import arrow.core.getOrElse
-import arrow.effects.IO
import arrow.syntax.function.curried
import org.apache.commons.cli.CommandLine
-import org.onap.dcae.collectors.veshv.utils.arrow.ExitFailure
+import org.onap.dcae.collectors.veshv.utils.process.ExitCode
+import org.onap.dcae.collectors.veshv.utils.process.ExitFailure
import org.onap.dcae.collectors.veshv.utils.arrow.fromNullablesChain
/**
val handleWrongArgumentErrorCurried = ::handleWrongArgumentError.curried()
-fun handleWrongArgumentError(programName: String, err: WrongArgumentError): IO<Unit> = IO {
+fun handleWrongArgumentError(programName: String, err: WrongArgumentError): ExitCode {
err.printMessage()
err.printHelp(programName)
-}.flatMap { ExitFailure(2).io() }
+ return ExitFailure(2)
+}
fun CommandLine.longValue(cmdLineOpt: CommandLineOption, default: Long): Long =
longValue(cmdLineOpt).getOrElse { default }
<version>${project.parent.version}</version>
<scope>compile</scope>
</dependency>
- <dependency>
- <groupId>${project.parent.groupId}</groupId>
- <artifactId>hv-collector-health-check</artifactId>
- <version>${project.parent.version}</version>
- </dependency>
<dependency>
<groupId>${project.parent.groupId}</groupId>
<artifactId>hv-collector-utils</artifactId>
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
-
<dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-reflect</artifactId>
+ <groupId>io.arrow-kt</groupId>
+ <artifactId>arrow-core-data</artifactId>
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
- <artifactId>arrow-core</artifactId>
+ <artifactId>arrow-extras-data</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
</dependency>
- <dependency>
- <groupId>io.projectreactor.addons</groupId>
- <artifactId>reactor-extra</artifactId>
- </dependency>
<dependency>
<groupId>io.projectreactor.netty</groupId>
<artifactId>reactor-netty</artifactId>
<groupId>io.projectreactor.kafka</groupId>
<artifactId>reactor-kafka</artifactId>
</dependency>
- <dependency>
- <groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId>
- <artifactId>cbs-client</artifactId>
- </dependency>
</dependencies>
</project>
</dependency>
<dependency>
<groupId>${project.parent.groupId}</groupId>
- <artifactId>hv-collector-xnf-simulator</artifactId>
+ <artifactId>hv-collector-ves-message-generator</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
*/
package org.onap.dcae.collectors.veshv.tests.fakes
-import arrow.effects.IO
import org.onap.dcae.collectors.veshv.boundary.Sink
import org.onap.dcae.collectors.veshv.model.ConsumedMessage
import org.onap.dcae.collectors.veshv.model.FailedToConsumeMessage
</profile>
</profiles>
<dependencies>
- <dependency>
- <groupId>${project.parent.groupId}</groupId>
- <artifactId>hv-collector-domain</artifactId>
- <version>${project.parent.version}</version>
- </dependency>
<dependency>
<groupId>${project.parent.groupId}</groupId>
<artifactId>hv-collector-commandline</artifactId>
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>io.arrow-kt</groupId>
- <artifactId>arrow-effects</artifactId>
- </dependency>
- <dependency>
- <groupId>io.arrow-kt</groupId>
- <artifactId>arrow-effects-instances</artifactId>
- </dependency>
- <dependency>
- <groupId>io.arrow-kt</groupId>
- <artifactId>arrow-effects-reactor</artifactId>
- </dependency>
- <dependency>
- <groupId>io.arrow-kt</groupId>
- <artifactId>arrow-syntax</artifactId>
- </dependency>
<dependency>
<groupId>io.projectreactor.kafka</groupId>
<artifactId>reactor-kafka</artifactId>
</dependency>
- <dependency>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java-util</artifactId>
- </dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
import arrow.core.getOrElse
import org.onap.dcae.collectors.veshv.utils.arrow.getOption
import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import reactor.core.publisher.Mono
import java.io.InputStream
-import java.lang.IllegalArgumentException
import java.util.concurrent.atomic.AtomicReference
/**
*/
package org.onap.dcae.collectors.veshv.simulators.dcaeapp.impl.adapters
-import arrow.effects.IO
import org.onap.dcae.collectors.veshv.simulators.dcaeapp.impl.DcaeAppSimulator
import org.onap.dcae.collectors.veshv.utils.NettyServerHandle
import org.onap.dcae.collectors.veshv.utils.ServerHandle
-import org.onap.dcae.collectors.veshv.utils.http.*
+import org.onap.dcae.collectors.veshv.utils.http.HttpConstants
+import org.onap.dcae.collectors.veshv.utils.http.HttpStatus
+import org.onap.dcae.collectors.veshv.utils.http.Response
+import org.onap.dcae.collectors.veshv.utils.http.Responses
+import org.onap.dcae.collectors.veshv.utils.http.sendAndHandleErrors
+import org.onap.dcae.collectors.veshv.utils.http.sendOrError
import org.onap.dcae.collectors.veshv.utils.logging.Logger
import reactor.core.publisher.Mono
import reactor.netty.http.server.HttpServer
)
}
- fun start(socketAddress: InetSocketAddress, kafkaTopics: Set<String>): IO<ServerHandle> =
- IO {
+ fun start(socketAddress: InetSocketAddress, kafkaTopics: Set<String>): Mono<ServerHandle> =
+ Mono.defer {
simulator.listenToTopics(kafkaTopics)
HttpServer.create()
.host(socketAddress.hostName)
.port(socketAddress.port)
.route(::setRoutes)
- .let { NettyServerHandle(it.bindNow()) }
+ .bind()
+ .map { NettyServerHandle(it) }
}
private fun setRoutes(route: HttpServerRoutes) {
req
.receive().aggregate().asString()
.flatMap {
- res.sendOrError{ simulator.listenToTopics(it) }
+ res.sendOrError { simulator.listenToTopics(it) }
}
}
.delete("/messages") { _, res ->
*/
package org.onap.dcae.collectors.veshv.simulators.dcaeapp
-import arrow.effects.IO
import org.onap.dcae.collectors.veshv.commandline.handleWrongArgumentErrorCurried
import org.onap.dcae.collectors.veshv.simulators.dcaeapp.impl.ConsumerFactory
import org.onap.dcae.collectors.veshv.simulators.dcaeapp.impl.DcaeAppSimulator
import org.onap.dcae.collectors.veshv.simulators.dcaeapp.impl.adapters.DcaeAppApiServer
import org.onap.dcae.collectors.veshv.simulators.dcaeapp.impl.config.ArgDcaeAppSimConfiguration
import org.onap.dcae.collectors.veshv.simulators.dcaeapp.impl.config.DcaeAppSimConfiguration
-import org.onap.dcae.collectors.veshv.utils.arrow.ExitFailure
-import org.onap.dcae.collectors.veshv.utils.arrow.unsafeRunEitherSync
import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import org.onap.dcae.collectors.veshv.utils.process.ExitCode
+import org.onap.dcae.collectors.veshv.utils.process.ExitSuccess
import org.onap.dcae.collectors.veshv.ves.message.generator.factory.MessageGeneratorFactory
private const val PACKAGE_NAME = "org.onap.dcae.collectors.veshv.simulators.dcaeapp"
private val logger = Logger(PACKAGE_NAME)
const val PROGRAM_NAME = "java $PACKAGE_NAME.MainKt"
-fun main(args: Array<String>) =
+fun main(args: Array<String>): Unit =
ArgDcaeAppSimConfiguration().parse(args)
- .mapLeft(handleWrongArgumentErrorCurried(PROGRAM_NAME))
- .map(::startApp)
- .unsafeRunEitherSync(
- { ex ->
- logger.withError { log("Failed to start a server", ex) }
- ExitFailure(1)
- },
- {
- logger.info { "Started DCAE-APP Simulator API server" }
- }
- )
+ .fold(handleWrongArgumentErrorCurried(PROGRAM_NAME), ::startApp)
+ .let(ExitCode::doExit)
-private fun startApp(config: DcaeAppSimConfiguration): IO<Unit> {
- logger.info { "Using configuration: $config" }
+
+private fun startApp(config: DcaeAppSimConfiguration): ExitSuccess {
+ logger.info { "Starting DCAE-APP Simulator API server with configuration: $config" }
val consumerFactory = ConsumerFactory(config.kafkaBootstrapServers)
val generatorFactory = MessageGeneratorFactory(config.maxPayloadSizeBytes)
val messageStreamValidation = MessageStreamValidation(generatorFactory.createVesEventGenerator())
- return DcaeAppApiServer(DcaeAppSimulator(consumerFactory, messageStreamValidation))
+ DcaeAppApiServer(DcaeAppSimulator(consumerFactory, messageStreamValidation))
.start(config.apiAddress, config.kafkaTopics)
.flatMap { it.await() }
+ .block()
+ return ExitSuccess
}
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
- <artifactId>arrow-core</artifactId>
+ <artifactId>arrow-core-data</artifactId>
</dependency>
<dependency>
<groupId>org.assertj</groupId>
<groupId>io.projectreactor.netty</groupId>
<artifactId>reactor-netty</artifactId>
</dependency>
- <dependency>
- <groupId>io.arrow-kt</groupId>
- <artifactId>arrow-effects</artifactId>
- </dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<scope>test</scope>
</dependency>
-
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-test</artifactId>
*/
package org.onap.dcae.collectors.veshv.healthcheck.factory
-import arrow.effects.IO
import io.netty.handler.codec.http.HttpResponseStatus
import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription
import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
private val healthDescription: AtomicReference<HealthDescription> = AtomicReference(HealthDescription.STARTING)
- fun start(): IO<ServerHandle> = IO {
+ fun start(): Mono<ServerHandle> = Mono.defer {
healthState().subscribe(healthDescription::set)
- val ctx = HttpServer.create()
+ HttpServer.create()
.tcpConfiguration {
it.addressSupplier { listenAddress }
.doOnUnbound { logClose() }
routes.get("/health/alive", ::livenessHandler)
routes.get("/monitoring/prometheus", ::monitoringHandler)
}
- NettyServerHandle(ctx.bindNow())
+ .bind()
+ .map { NettyServerHandle(it) }
+
}
private fun readinessHandler(_req: HttpServerRequest, resp: HttpServerResponse) =
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
- <artifactId>arrow-core</artifactId>
+ <artifactId>arrow-core-data</artifactId>
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
}
}
- HealthCheckServer.start(configurationModule.healthCheckPort(args))
+ HealthCheckServer.start(configurationModule.healthCheckPort(args)).block()
configurationModule
.hvVesConfigurationUpdates(args, configStateListener, ServiceContext::mdc)
.publishOn(Schedulers.single(Schedulers.elastic()))
import org.onap.dcae.collectors.veshv.main.metrics.MicrometerMetrics
import org.onap.dcae.collectors.veshv.model.ServiceContext
import org.onap.dcae.collectors.veshv.utils.ServerHandle
-import org.onap.dcae.collectors.veshv.utils.arrow.then
import org.onap.dcae.collectors.veshv.utils.logging.Logger
import java.net.InetSocketAddress
fun start(port: Int) =
createHealthCheckServer(port)
.start()
- .then(::logServerStarted)
- .unsafeRunSync()
+ .doOnSuccess(::logServerStarted)
private fun createHealthCheckServer(listenPort: Int) =
HealthCheckApiServer(
<dependency>
<groupId>io.arrow-kt</groupId>
- <artifactId>arrow-core</artifactId>
- </dependency>
- <dependency>
- <groupId>io.arrow-kt</groupId>
- <artifactId>arrow-syntax</artifactId>
+ <artifactId>arrow-core-data</artifactId>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>kotlin-reflect</artifactId>
</dependency>
<dependency>
- <groupId>io.arrow-kt</groupId>
- <artifactId>arrow-instances-data</artifactId>
+ <groupId>org.jetbrains.kotlin</groupId>
+ <artifactId>kotlin-stdlib-jdk8</artifactId>
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
- <artifactId>arrow-effects</artifactId>
+ <artifactId>arrow-typeclasses</artifactId>
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
- <artifactId>arrow-effects-instances</artifactId>
+ <artifactId>arrow-core-extensions</artifactId>
</dependency>
<dependency>
<groupId>io.arrow-kt</groupId>
<artifactId>arrow-syntax</artifactId>
</dependency>
- <dependency>
- <groupId>org.jetbrains.kotlinx</groupId>
- <artifactId>kotlinx-coroutines-core</artifactId>
- </dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
import arrow.core.ForOption
import arrow.core.Option
import arrow.core.Try
+import arrow.core.extensions.option.monad.monad
import arrow.core.fix
import arrow.core.identity
-import arrow.effects.ForIO
-import arrow.effects.IO
-import arrow.effects.fix
-import arrow.effects.instances.io.monad.monad
-import arrow.instances.option.monad.monad
import arrow.syntax.collections.firstOption
import arrow.typeclasses.MonadContinuation
-import arrow.typeclasses.binding
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.util.concurrent.atomic.AtomicReference
: Option<A> = Option.monad().binding(c).fix()
}
-object IOUtils {
- fun <A> binding(c: suspend MonadContinuation<ForIO, *>.() -> A)
- : IO<A> = IO.monad().binding(c).fix()
-}
-
fun <A> Either<A, A>.flatten() = fold(::identity, ::identity)
fun <B> Either<Throwable, B>.rightOrThrow() = fold({ throw it }, ::identity)
object Entry : Marker(ENTRY)
object Exit : Marker(EXIT)
- class Invoke(id: UUID = UUID.randomUUID(), timestamp: Instant = Instant.now()) : Marker(INVOKE, mdc(id, timestamp)) {
+ class Invoke(id: UUID = UUID.randomUUID(), timestamp: Instant = Instant.now()) :
+ Marker(INVOKE, mdc(id, timestamp)) {
companion object {
private fun mdc(id: UUID, timestamp: Instant) = mapOf(
OnapMdc.INVOCATION_ID to id.toString(),
* limitations under the License.
* ============LICENSE_END=========================================================
*/
-package org.onap.dcae.collectors.veshv.utils.arrow
+package org.onap.dcae.collectors.veshv.utils.process
-import arrow.core.Either
-import arrow.core.Left
-import arrow.core.Right
-import arrow.effects.IO
-import org.reactivestreams.Publisher
-import reactor.core.publisher.Flux
-import reactor.core.publisher.Mono
-import reactor.core.publisher.toMono
import kotlin.system.exitProcess
/**
sealed class ExitCode {
abstract val code: Int
- fun io() = IO {
- exitProcess(code)
- }
+ fun doExit(): Nothing = exitProcess(code)
+
}
object ExitSuccess : ExitCode() {
}
data class ExitFailure(override val code: Int) : ExitCode()
-
-inline fun <A, B> Either<IO<A>, IO<B>>.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> 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(
- { Flux.error<T>(it) },
- { Flux.just<T>(it) }
- )
- }
-
-inline fun <T> IO<T>.then(crossinline block: (T) -> Unit): IO<T> =
- map {
- block(it)
- it
- }
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018-2019 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=========================================================
+ */
+
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+package org.onap.dcae.collectors.veshv.utils.rx
+
+import org.reactivestreams.Publisher
+import reactor.core.publisher.Mono
+import reactor.core.publisher.toMono
+
+fun <T> Publisher<T>.then(callback: () -> Unit): Mono<Unit> =
+ toMono().then(Mono.fromCallable(callback))
*/
package org.onap.dcae.collectors.veshv.utils
-import arrow.effects.IO
import org.onap.dcae.collectors.veshv.utils.logging.Logger
import reactor.core.publisher.Mono
import reactor.netty.DisposableServer
* @since August 2018
*/
abstract class ServerHandle(val host: String, val port: Int) : Closeable {
- abstract fun await(): IO<Unit>
+ abstract fun await(): Mono<Void>
}
/**
}
}
- override fun await() = IO<Unit> {
- ctx.channel().closeFuture().sync()
+ override fun await(): Mono<Void> = Mono.create { callback ->
+ ctx.channel().closeFuture().addListener {
+ callback.success()
+ }
}
companion object {
</profiles>
<dependencies>
- <dependency>
- <groupId>${project.parent.groupId}</groupId>
- <artifactId>hv-collector-domain</artifactId>
- <version>${project.parent.version}</version>
- </dependency>
<dependency>
<groupId>${project.parent.groupId}</groupId>
<artifactId>hv-collector-ssl</artifactId>
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>io.arrow-kt</groupId>
- <artifactId>arrow-effects</artifactId>
- </dependency>
- <dependency>
- <groupId>io.arrow-kt</groupId>
- <artifactId>arrow-effects-instances</artifactId>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlinx</groupId>
- <artifactId>kotlinx-coroutines-core</artifactId>
- </dependency>
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-stdlib-jdk8</artifactId>
<artifactId>logback-classic</artifactId>
<scope>runtime</scope>
</dependency>
- <!-- See comment in main pom
- <dependency>
- <groupId>io.netty</groupId>
- <artifactId>netty-transport-native-epoll</artifactId>
- <classifier>${os.detected.classifier}</classifier>
- </dependency>
- <dependency>
- <groupId>io.netty</groupId>
- <artifactId>netty-tcnative-boringssl-static</artifactId>
- <classifier>${os.detected.classifier}</classifier>
- </dependency>
- -->
<dependency>
<groupId>org.glassfish</groupId>
<artifactId>javax.json</artifactId>
import arrow.core.Either
import arrow.core.Some
import arrow.core.Try
+import arrow.core.extensions.either.monad.monad
import arrow.core.fix
-import arrow.effects.IO
-import arrow.instances.either.monad.monad
-import arrow.typeclasses.binding
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.adapters.HvVesClient
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.factory.ClientFactory
-import org.onap.dcae.collectors.veshv.utils.arrow.asIo
import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParameters
import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParametersParser
import org.onap.dcae.collectors.veshv.ves.message.generator.api.ParsingError
private val defaultHvVesClient by lazy { clientFactory.create() }
- fun startSimulation(messageParameters: InputStream): Either<ParsingError, IO<Unit>> =
+ fun startSimulation(messageParameters: InputStream): Either<ParsingError, Mono<Void>> =
Either.monad<ParsingError>().binding {
val json = parseJsonArray(messageParameters).bind()
val parameters = messageParametersParser.parse(json).bind()
.mapLeft { ParsingError("Failed to parse JSON", Some(it)) }
- private fun simulationFrom(parameters: List<MessageParameters>): IO<Unit> =
+ private fun simulationFrom(parameters: List<MessageParameters>): Mono<Void> =
parameters
.map(::asClientToMessages)
.groupMessagesByClients()
.toList()
.toFlux()
.map(::simulate)
- .then(Mono.just(Unit))
- .asIo()
+ .then()
private fun <M> List<Pair<HvVesClient, M>>.groupMessagesByClients() =
groupBy({ it.first }, { it.second })
*/
package org.onap.dcae.collectors.veshv.simulators.xnf.impl.adapters
-import org.onap.dcae.collectors.veshv.utils.arrow.then
import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import org.onap.dcae.collectors.veshv.utils.rx.then
import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducer
import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.options.PayloadType
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.nio.ByteBuffer
-import java.util.concurrent.atomic.AtomicLong
/**
* @author Jakub Dudycz <jakub.dudycz@nokia.com>
package org.onap.dcae.collectors.veshv.simulators.xnf.impl.adapters
import arrow.core.Either
-import arrow.effects.IO
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.OngoingSimulations
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.XnfSimulator
import org.onap.dcae.collectors.veshv.utils.NettyServerHandle
private val xnfSimulator: XnfSimulator,
private val ongoingSimulations: OngoingSimulations) {
- fun start(socketAddress: InetSocketAddress): IO<ServerHandle> = IO {
- HttpServer.create()
- .host(socketAddress.hostName)
- .port(socketAddress.port)
- .route(::setRoutes)
- .let { NettyServerHandle(it.bindNow()) }
- }
+ fun start(socketAddress: InetSocketAddress): Mono<ServerHandle> =
+ HttpServer.create()
+ .host(socketAddress.hostName)
+ .port(socketAddress.port)
+ .route(::setRoutes)
+ .bind()
+ .map { NettyServerHandle(it) }
+
private fun setRoutes(route: HttpServerRoutes) {
route
internal class XnfHealthCheckServer {
fun startServer(config: SimulatorConfiguration) = createHealthCheckServer(config)
.start()
- .map { logger.info(serverStartedMessage(it)); it }
+ .doOnNext { logger.info(serverStartedMessage(it)) }
private fun createHealthCheckServer(config: SimulatorConfiguration): HealthCheckApiServer {
val monitoring = object : PrometheusMetricsProvider {
*/
package org.onap.dcae.collectors.veshv.simulators.xnf.impl
-import arrow.effects.IO
-import kotlinx.coroutines.asCoroutineDispatcher
import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription.BUSY
import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription.IDLE
import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.adapters.XnfApiServer
import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Scheduler
+import reactor.core.scheduler.Schedulers
import java.util.*
import java.util.concurrent.ConcurrentHashMap
-import java.util.concurrent.Executor
-import java.util.concurrent.Executors
/**
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
* @since August 2018
*/
-class OngoingSimulations(executor: Executor = Executors.newCachedThreadPool(),
+class OngoingSimulations(private val scheduler: Scheduler = Schedulers.elastic(),
private val healthState: HealthState = HealthState.INSTANCE) {
- private val asyncSimulationContext = executor.asCoroutineDispatcher()
private val simulations = ConcurrentHashMap<UUID, Status>()
- fun startAsynchronousSimulation(simulationIo: IO<Unit>): UUID {
+ fun startAsynchronousSimulation(simulationIo: Mono<Void>): UUID {
val id = UUID.randomUUID()
simulations[id] = StatusOngoing
updateHealthState()
- simulationIo.continueOn(asyncSimulationContext).unsafeRunAsync { result ->
- result.fold(
- { err ->
- logger.withWarn { log("Error", err) }
- simulations[id] = StatusFailure(err)
- },
- {
- logger.info { "Finished sending messages" }
- simulations[id] = StatusSuccess
- }
- ).also { updateHealthState() }
- }
+ simulationIo
+ .publishOn(scheduler)
+ .doOnSuccess {
+ logger.info { "Finished sending messages" }
+ simulations[id] = StatusSuccess
+ }
+ .doOnError { err ->
+ logger.withWarn { log("Error", err) }
+ simulations[id] = StatusFailure(err)
+ }
+ .doFinally { updateHealthState() }
+ .subscribe()
+
return id
}
*/
package org.onap.dcae.collectors.veshv.simulators.xnf
-import arrow.effects.IO
import io.vavr.collection.HashSet
+import org.onap.dcae.collectors.veshv.commandline.handleWrongArgumentError
import org.onap.dcae.collectors.veshv.commandline.handleWrongArgumentErrorCurried
import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription
import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.config.ClientConfiguration
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.config.SimulatorConfiguration
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.factory.ClientFactory
-import org.onap.dcae.collectors.veshv.utils.arrow.ExitFailure
-import org.onap.dcae.collectors.veshv.utils.arrow.IOUtils.binding
-import org.onap.dcae.collectors.veshv.utils.arrow.unsafeRunEitherSync
+import org.onap.dcae.collectors.veshv.utils.process.ExitCode
+import org.onap.dcae.collectors.veshv.utils.process.ExitSuccess
import org.onap.dcae.collectors.veshv.utils.logging.Logger
import org.onap.dcae.collectors.veshv.ves.message.generator.factory.MessageGeneratorFactory
* @author Jakub Dudycz <jakub.dudycz@nokia.com>
* @since June 2018
*/
-fun main(args: Array<String>) = ArgXnfSimulatorConfiguration().parse(args)
- .mapLeft(handleWrongArgumentErrorCurried(PROGRAM_NAME))
- .map(::startServers)
- .unsafeRunEitherSync(
- { ex ->
- logger.withError { log("Failed to start a server", ex) }
- ExitFailure(1)
- },
- {
- logger.info { "Stopping xNF Simulator API server" }
- }
- )
+fun main(args: Array<String>): Unit =
+ ArgXnfSimulatorConfiguration().parse(args)
+ .fold(handleWrongArgumentErrorCurried(PROGRAM_NAME), ::startServers)
+ .let(ExitCode::doExit)
-private fun startServers(config: SimulatorConfiguration): IO<Unit> =
- binding {
- logger.info { "Using configuration: $config" }
+private fun startServers(config: SimulatorConfiguration): ExitCode {
+ logger.info { "Using configuration: $config" }
- XnfHealthCheckServer().startServer(config).bind()
+ XnfHealthCheckServer().startServer(config).block()
- val clientConfig = ClientConfiguration(HashSet.of(config.hvVesAddress), config.securityProvider)
- val xnfSimulator = XnfSimulator(
- ClientFactory(clientConfig),
- MessageGeneratorFactory(config.maxPayloadSizeBytes)
- )
- val xnfApiServerHandler = XnfApiServer(xnfSimulator, OngoingSimulations())
- .start(config.listenAddress).bind()
+ val clientConfig = ClientConfiguration(HashSet.of(config.hvVesAddress), config.securityProvider)
+ val xnfSimulator = XnfSimulator(
+ ClientFactory(clientConfig),
+ MessageGeneratorFactory(config.maxPayloadSizeBytes)
+ )
+ val xnfApiServerHandler = XnfApiServer(xnfSimulator, OngoingSimulations())
+ .start(config.listenAddress)
+ .block()
- logger.info { "Started xNF Simulator API server" }
- HealthState.INSTANCE.changeState(HealthDescription.IDLE)
+ logger.info { "Started xNF Simulator API server" }
+ HealthState.INSTANCE.changeState(HealthDescription.IDLE)
- xnfApiServerHandler.await().bind()
- }
+ xnfApiServerHandler.await().block()
+ return ExitSuccess
+}
*/
package org.onap.dcae.collectors.veshv.main
-import arrow.effects.IO
import org.assertj.core.api.Assertions.assertThat
import org.jetbrains.spek.api.Spek
import org.jetbrains.spek.api.dsl.describe
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.StatusOngoing
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.StatusSuccess
import org.onap.dcae.collectors.veshv.tests.utils.waitUntilSucceeds
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
import java.util.*
-import java.util.concurrent.Executors
/**
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
* @since September 2018
*/
internal class OngoingSimulationsTest : Spek({
- val executor = Executors.newSingleThreadExecutor()
- val cut = OngoingSimulations(executor)
+ val scheduler = Schedulers.single()
+ val cut = OngoingSimulations(scheduler)
describe("simulations repository") {
given("not existing task task id") {
}
afterGroup {
- executor.shutdown()
+ scheduler.dispose()
}
}
afterEachTest { cut.clear() }
})
-private fun neverendingTask() = IO.async<Unit> { }
+private fun neverendingTask() = Mono.never<Void>()
-private fun succesfulTask(): IO<Unit> = IO { println("great success!") }
+private fun succesfulTask(): Mono<Void> = Mono.empty<Void>()
+ .doOnSuccess {
+ println("great success")
+ }
-private fun failingTask(): Pair<RuntimeException, IO<Unit>> {
+private fun failingTask(): Pair<RuntimeException, Mono<Void>> {
val cause = RuntimeException("facepalm")
- val task = IO.raiseError<Unit>(cause)
+ val task = Mono.error<Void>(cause)
return Pair(cause, task)
}
whenever(vesClient.sendRawPayload(any(), eq(PayloadType.PROTOBUF))).thenReturn(Mono.just(Unit))
// when
- cut.startSimulation(json).map { it.unsafeRunSync() }
+ cut.startSimulation(json).map { it.block() }
// then
verify(vesClient).sendRawPayload(any(), eq(PayloadType.PROTOBUF))
Code Review / dcaegen2 / collectors / hv-ves.git / commitdiff