DOMAIN=${1:-perf3gpp}
TOPIC=${2:-HV_VES_PERF3GPP}
-CONFIGURATION="
-{
- \"collector.routing\":
- [{
- \"fromDomain\": \"${DOMAIN}\",
- \"toTopic\": \"${TOPIC}\"
- }]
+CONFIGURATION="{
+ "streams_publishes": {
+ "${DOMAIN}": {
+ "type": "kafka",
+ "kafka_info": {
+ "bootstrap_servers": "message-router-kafka:9092",
+ "topic_name": "${TOPIC}"
+ }
+ }
+ }
}"
CONFIGURATION_ENDPOINT=localhost:8500/v1/kv/veshv-config
image: docker.io/consul:1.0.6
restart: on-failure
command: ["kv", "put", "-http-addr=http://consul-server:8500", "dcae-hv-ves-collector", '{
- "collector.routing": [
- {
- "fromDomain": "perf3gpp",
- "toTopic": "HV_VES_PERF3GPP"
+ "streams_publishes": {
+ "perf3gpp": {
+ "type": "kafka",
+ "kafka_info": {
+ "bootstrap_servers": "message-router-kafka:9092",
+ "topic_name": "HV_VES_PERF3GPP"
+ }
}
- ]
+ }
}'
]
depends_on:
- "6061:6061/tcp"
environment:
JAVA_OPTS: "-Dio.netty.leakDetection.level=paranoid -Dlogback.configurationFile=/etc/ONAP/dcae-hv-ves-collector/logback.xml"
+ VESHV_CONFIGURATION_FILE: "/etc/ves-hv/configuration/base.json"
CONSUL_HOST: "consul-server"
CONFIG_BINDING_SERVICE: "CBS"
HOSTNAME: "dcae-hv-ves-collector"
*/
package org.onap.dcae.collectors.veshv.config.api
+import arrow.core.getOrElse
import org.onap.dcae.collectors.veshv.config.api.model.HvVesConfiguration
import org.onap.dcae.collectors.veshv.config.api.model.MissingArgumentException
import org.onap.dcae.collectors.veshv.config.api.model.ValidationException
import org.onap.dcae.collectors.veshv.config.impl.ConfigurationValidator
import org.onap.dcae.collectors.veshv.config.impl.FileConfigurationReader
import org.onap.dcae.collectors.veshv.config.impl.HvVesCommandLineParser
-import org.onap.dcae.collectors.veshv.utils.arrow.rightOrThrow
+import org.onap.dcae.collectors.veshv.config.impl.PartialConfiguration
import org.onap.dcae.collectors.veshv.utils.arrow.throwOnLeft
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
import org.onap.dcae.collectors.veshv.utils.logging.MappedDiagnosticContext
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.EnvProperties
import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
class ConfigurationModule {
private val cmd = HvVesCommandLineParser()
private val configReader = FileConfigurationReader()
private val configValidator = ConfigurationValidator()
+ private val merger = ConfigurationMerger()
fun healthCheckPort(args: Array<String>): Int = cmd.getHealthcheckPort(args)
fun hvVesConfigurationUpdates(args: Array<String>,
configStateListener: ConfigurationStateListener,
mdc: MappedDiagnosticContext): Flux<HvVesConfiguration> =
- Flux.just(cmd.getConfigurationFile(args))
- .throwOnLeft { MissingArgumentException(it.message, it.cause) }
- .map { it.reader().use(configReader::loadConfig) }
+ Mono.just(cmd.getConfigurationFile(args))
+ .throwOnLeft(::MissingArgumentException)
+ .map {
+ logger.info { "Using base configuration file: ${it.absolutePath}" }
+ it.reader().use(configReader::loadConfig)
+ }
.cache()
- .flatMap { basePartialConfig ->
- val baseConfig = configValidator.validate(basePartialConfig)
- .rightOrThrow { ValidationException(it.message) }
- val cbsConfigProvider = CbsConfigurationProvider(
- CbsClientFactory.createCbsClient(EnvProperties.fromEnvironment()),
- baseConfig.cbs,
- configStateListener,
- mdc)
- val merger = ConfigurationMerger()
- cbsConfigProvider()
+ .flatMapMany { basePartialConfig ->
+ cbsConfigurationProvider(basePartialConfig, configStateListener, mdc)
+ .invoke()
.map { merger.merge(basePartialConfig, it) }
- .map { configValidator.validate(it) }
- .throwOnLeft { ValidationException(it.message) }
+ .map(configValidator::validate)
+ .throwOnLeft()
}
+ private fun cbsConfigurationProvider(basePartialConfig: PartialConfiguration,
+ configStateListener: ConfigurationStateListener,
+ mdc: MappedDiagnosticContext): CbsConfigurationProvider =
+ CbsConfigurationProvider(
+ CbsClientFactory.createCbsClient(EnvProperties.fromEnvironment()),
+ cbsConfigurationFrom(basePartialConfig),
+ configStateListener,
+ mdc)
+
+ private fun cbsConfigurationFrom(basePartialConfig: PartialConfiguration) =
+ configValidator.validatedCbsConfiguration(basePartialConfig)
+ .getOrElse { throw ValidationException("Invalid CBS section defined in configuration file") }
+
+ companion object {
+ private val logger = Logger(ConfigurationModule::class)
+ }
}
*/
package org.onap.dcae.collectors.veshv.config.api.model
-class MissingArgumentException(message: String, cause: Throwable?) : RuntimeException(message, cause)
+import org.onap.dcae.collectors.veshv.commandline.WrongArgumentError
+
+class MissingArgumentException(err: WrongArgumentError) : RuntimeException(err.message, err.cause)
class ValidationException(message: String) : RuntimeException(message)
updateOption: Option<PartialCollectorConfig>) =
applyUpdate(baseOption, updateOption) { base, update ->
PartialCollectorConfig(
- base.maxRequestSizeBytes.updateToGivenOrNone(update.maxRequestSizeBytes),
- base.kafkaServers.updateToGivenOrNone(update.kafkaServers),
base.routing.updateToGivenOrNone(update.routing)
)
}
import org.onap.dcae.collectors.veshv.config.api.model.CollectorConfiguration
import org.onap.dcae.collectors.veshv.config.api.model.HvVesConfiguration
import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration
+import org.onap.dcae.collectors.veshv.config.api.model.ValidationException
import org.onap.dcae.collectors.veshv.ssl.boundary.SecurityConfiguration
import org.onap.dcae.collectors.veshv.utils.arrow.OptionUtils.binding
import org.onap.dcae.collectors.veshv.utils.arrow.mapBinding
+import org.onap.dcae.collectors.veshv.utils.arrow.doOnEmpty
import org.onap.dcae.collectors.veshv.utils.logging.LogLevel
import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import java.net.InetSocketAddress
-import java.time.Duration
/**
* @author Jakub Dudycz <jakub.dudycz@nokia.com>
*/
internal class ConfigurationValidator {
- fun validate(partialConfig: PartialConfiguration)
- : Either<ValidationError, HvVesConfiguration> = binding {
- val logLevel = determineLogLevel(partialConfig.logLevel)
+ fun validate(partialConfig: PartialConfiguration) =
+ logger.info { "About to validate configuration: $partialConfig" }.let {
+ binding {
+ val logLevel = determineLogLevel(partialConfig.logLevel)
- val serverConfiguration = partialConfig.server.bind()
- .let { createServerConfiguration(it).bind() }
+ val serverConfiguration = validatedServerConfiguration(partialConfig)
+ .doOnEmpty { logger.debug { "Cannot bind server configuration" } }
+ .bind()
- val cbsConfiguration = partialConfig.cbs.bind()
- .let { createCbsConfiguration(it).bind() }
+ val cbsConfiguration = validatedCbsConfiguration(partialConfig)
+ .doOnEmpty { logger.debug { "Cannot bind cbs configuration" } }
+ .bind()
- val securityConfiguration = SecurityConfiguration(partialConfig.security.bind().keys)
+ val securityConfiguration = SecurityConfiguration(partialConfig.security.bind().keys)
-// TOD0: retrieve when ConfigurationMerger is implemented
-// val collectorConfiguration = partialConfig.collector.bind()
-// .let { createCollectorConfig(it).bind() }
+ val collectorConfiguration = validatedCollectorConfig(partialConfig)
+ .doOnEmpty { logger.debug { "Cannot bind collector configuration" } }
+ .bind()
+
+ HvVesConfiguration(
+ serverConfiguration,
+ cbsConfiguration,
+ securityConfiguration,
+ collectorConfiguration,
+ logLevel
+ )
+ }.toEither { ValidationException("Some required configuration options are missing") }
+ }
- HvVesConfiguration(
- serverConfiguration,
- cbsConfiguration,
- securityConfiguration,
-// TOD0: swap when ConfigurationMerger is implemented
-// collectorConfiguration
- CollectorConfiguration(emptyList()),
-// end TOD0
- logLevel
- )
- }.toEither { ValidationError("Some required configuration options are missing") }
private fun determineLogLevel(logLevel: Option<LogLevel>) =
logLevel.getOrElse {
DEFAULT_LOG_LEVEL
}
- private fun createServerConfiguration(partial: PartialServerConfig) =
+ private fun validatedServerConfiguration(partial: PartialConfiguration) =
partial.mapBinding {
- ServerConfiguration(
- it.listenPort.bind(),
- it.idleTimeoutSec.bind(),
- it.maxPayloadSizeBytes.bind()
- )
+ partial.server.bind().let {
+ ServerConfiguration(
+ it.listenPort.bind(),
+ it.idleTimeoutSec.bind(),
+ it.maxPayloadSizeBytes.bind()
+ )
+ }
}
- private fun createCbsConfiguration(partial: PartialCbsConfig) =
+ fun validatedCbsConfiguration(partial: PartialConfiguration) =
partial.mapBinding {
- CbsConfiguration(
- it.firstRequestDelaySec.bind(),
- it.requestIntervalSec.bind()
- )
+ it.cbs.bind().let {
+ CbsConfiguration(
+ it.firstRequestDelaySec.bind(),
+ it.requestIntervalSec.bind()
+ )
+ }
}
-// TOD0: retrieve when ConfigurationMerger is implemented
-// private fun createCollectorConfig(partial: PartialCollectorConfig) =
-// partial.mapBinding {
-// CollectorConfiguration(
-// it.maxRequestSizeBytes.bind(),
-// toKafkaServersString(it.kafkaServers.bind()),
-// it.routing.bind()
-// )
-// }
-
- private fun toKafkaServersString(kafkaServers: List<InetSocketAddress>): String =
- kafkaServers.joinToString(",") { "${it.hostName}:${it.port}" }
+ private fun validatedCollectorConfig(partial: PartialConfiguration) =
+ partial.mapBinding {
+ partial.collector.bind().let {
+ CollectorConfiguration(
+ it.routing.bind()
+ )
+ }
+ }
companion object {
val DEFAULT_LOG_LEVEL = LogLevel.INFO
private val logger = Logger(ConfigurationValidator::class)
}
}
-
-data class ValidationError(val message: String, val cause: Option<Throwable> = None)
import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.DurationOfSecondsAdapter
import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.OptionAdapter
import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.SecurityAdapter
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
import java.io.Reader
import java.time.Duration
fun loadConfig(input: Reader): PartialConfiguration =
gson.fromJson(input, PartialConfiguration::class.java)
+ .also { logger.info { "Successfully read file and parsed json to configuration: $it" } }
+
+ companion object {
+ private val logger = Logger(FileConfigurationReader::class)
+ }
}
it.stringValue(CONFIGURATION_FILE).map(::File)
}.toEither {
WrongArgumentError(
- message = "Unexpected error when parsing command line arguments",
+ message = "Base configuration filepath missing on command line",
cmdLineOptionsList = cmdLineOptionsList)
}
parse(args) {
it.intValue(HEALTH_CHECK_API_PORT)
}.getOrElse {
- logger.info { "Healthcheck port missing on command line," +
- " using default: $DEFAULT_HEALTHCHECK_PORT" }
+ logger.info { "Healthcheck port missing on command line, using default: $DEFAULT_HEALTHCHECK_PORT" }
DEFAULT_HEALTHCHECK_PORT
}
internal data class PartialSecurityConfig(val keys: Option<SecurityKeys> = None)
internal data class PartialCollectorConfig(
- val maxRequestSizeBytes: Option<Int> = None,
- val kafkaServers: Option<List<InetSocketAddress>> = None, // TOD0: remove properties and simplify this part
val routing: Option<Routing> = None
)
Some(mock())
)),
Some(PartialCollectorConfig(
- Some(4),
- Some(emptyList()),
someFromEmptyRouting
)),
None
securityKeys
)),
Some(PartialCollectorConfig(
- Some(4),
- Some(emptyList()),
someFromEmptyRouting
)),
Some(LogLevel.INFO)
securityKeys
)),
Some(PartialCollectorConfig(
- Some(4),
- Some(emptyList()),
someFromEmptyRouting
)),
Some(LogLevel.INFO)
.transform(::filterInvalidWireFrame)
.transform(::decodeProtobufPayload)
.transform(::filterInvalidProtobufMessages)
- // TOD0: try to remove new flux creation in Sink interface to avoid two calls to handleErrors here
- .handleErrors()
.transform(::route)
.handleErrors()
.doFinally { releaseBuffersMemory() }
}
}
+ private fun releaseBuffersMemory() = wireChunkDecoder.release()
+ .also { logger.debug { "Released buffer memory after handling message stream" } }
+
private fun <T> Flux<T>.handleErrors(): Flux<T> = onErrorResume {
metrics.notifyClientRejected(ClientRejectionCause.fromThrowable(it))
logger.handleReactiveStreamError(clientContext, it)
}
- private fun releaseBuffersMemory() = wireChunkDecoder.release()
- .also { logger.debug { "Released buffer memory after handling message stream" } }
-
private fun <T> Flux<T>.filterFailedWithLog(predicate: (T) -> MessageEither): Flux<T> =
filterFailedWithLog(logger, clientContext::fullMdc, predicate)
import org.onap.dcae.collectors.veshv.model.MessageDropCause.INVALID_MESSAGE
import org.onap.dcae.collectors.veshv.model.MessageDropCause.KAFKA_FAILURE
import org.onap.dcae.collectors.veshv.model.MessageDropCause.ROUTE_NOT_FOUND
+import org.onap.dcae.collectors.veshv.tests.component.Sut.Companion.MAX_PAYLOAD_SIZE_BYTES
import org.onap.dcae.collectors.veshv.tests.fakes.ALTERNATE_PERF3GPP_TOPIC
import org.onap.dcae.collectors.veshv.tests.fakes.PERF3GPP_TOPIC
import org.onap.dcae.collectors.veshv.tests.fakes.basicRouting
.isEqualTo(1)
}
- it("should gather metrics for sing errors") {
+ it("should gather metrics for sink errors") {
val sut = vesHvWithAlwaysFailingSink(basicRouting)
sut.handleConnection(vesWireFrameMessage(domain = PERF3GPP))
given("rejection causes") {
mapOf(
ClientRejectionCause.PAYLOAD_SIZE_EXCEEDED_IN_MESSAGE to
- messageWithPayloadOfSize(Sut.MAX_PAYLOAD_SIZE_BYTES + 1),
+ messageWithPayloadOfSize(MAX_PAYLOAD_SIZE_BYTES + 1),
ClientRejectionCause.INVALID_WIRE_FRAME_MARKER to garbageFrame()
).forEach { cause, vesMessage ->
on("cause $cause") {
override fun close() = collectorProvider.close()
companion object {
- const val MAX_PAYLOAD_SIZE_BYTES = 1024
+ const val MAX_PAYLOAD_SIZE_BYTES = 1024 * 1024
}
}
import org.assertj.core.api.Assertions.assertThat
import org.jetbrains.spek.api.Spek
import org.jetbrains.spek.api.dsl.describe
-import org.jetbrains.spek.api.dsl.given
import org.jetbrains.spek.api.dsl.it
import org.onap.dcae.collectors.veshv.config.api.model.CollectorConfiguration
import org.onap.dcae.collectors.veshv.config.api.model.Routing
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.MEASUREMENT
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.OTHER
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.PERF3GPP
-import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription
+import org.onap.dcae.collectors.veshv.tests.component.Sut.Companion.MAX_PAYLOAD_SIZE_BYTES
import org.onap.dcae.collectors.veshv.tests.fakes.ALTERNATE_PERF3GPP_TOPIC
import org.onap.dcae.collectors.veshv.tests.fakes.PERF3GPP_TOPIC
import org.onap.dcae.collectors.veshv.tests.fakes.StoringSink
import org.onap.dcae.collectors.veshv.tests.fakes.basicRouting
-import org.onap.dcae.collectors.veshv.tests.fakes.alternativeRouting
-import org.onap.dcae.collectors.veshv.tests.fakes.emptyRouting
import org.onap.dcae.collectors.veshv.tests.fakes.twoDomainsToOneTopicRouting
import org.onap.dcae.collectors.veshv.tests.utils.garbageFrame
import org.onap.dcae.collectors.veshv.tests.utils.messageWithInvalidWireFrameHeader
import org.onap.dcae.collectors.veshv.tests.utils.messageWithPayloadOfSize
import org.onap.dcae.collectors.veshv.tests.utils.vesWireFrameMessage
import org.onap.dcae.collectors.veshv.tests.utils.wireFrameMessageWithInvalidPayload
-import reactor.core.publisher.Flux
-import java.time.Duration
/**
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
val (sut, sink) = vesHvWithStoringSink()
val validMessage = vesWireFrameMessage(PERF3GPP)
val msgWithInvalidFrame = messageWithInvalidWireFrameHeader()
- val msgWithTooBigPayload = messageWithPayloadOfSize(Sut.MAX_PAYLOAD_SIZE_BYTES + 1, PERF3GPP)
+ val msgWithTooBigPayload = messageWithPayloadOfSize(MAX_PAYLOAD_SIZE_BYTES + 1, PERF3GPP)
val expectedRefCnt = 0
val handledEvents = sut.handleConnection(
val handledMessages = sut.handleConnection(sink,
vesWireFrameMessage(PERF3GPP, "first"),
- messageWithPayloadOfSize(Sut.MAX_PAYLOAD_SIZE_BYTES + 1, PERF3GPP),
+ messageWithPayloadOfSize(MAX_PAYLOAD_SIZE_BYTES + 1, PERF3GPP),
vesWireFrameMessage(PERF3GPP))
assertThat(handledMessages).hasSize(1)
const val PERF3GPP_TOPIC = "HV_VES_PERF3GPP"
const val ALTERNATE_PERF3GPP_TOPIC = "HV_VES_PERF3GPP_ALTERNATIVE"
const val KAFKA_BOOTSTRAP_SERVERS = "kafka:9092"
-const val MAX_PAYLOAD_SIZE_BYTES = 1024 * 1024
+private const val MAX_PAYLOAD_SIZE_BYTES = 1024 * 1024
private val perf3gppKafkaSink = ImmutableKafkaSink.builder()
.name("PERF3GPP")
FROM docker.io/openjdk:11-jre-slim
-LABEL copyright="Copyright (C) 2018 NOKIA"
+LABEL copyright="Copyright (C) 2018-2019 NOKIA"
LABEL license.name="The Apache Software License, Version 2.0"
LABEL license.url="http://www.apache.org/licenses/LICENSE-2.0"
LABEL maintainer="Nokia Wroclaw ONAP Team"
COPY src/main/docker/*.sh ./
COPY src/main/docker/base.json /etc/ves-hv/configuration/base.json
-COPY target/hv-collector-main-*.jar ./
+COPY target/hv-collector-main-*.jar ./
\ No newline at end of file
"requestIntervalSec": 5
},
"security": {
- },
- "collector": {
- "maxRequestSizeBytes": 1048576,
- "kafkaServers": [
- "message-router-kafka:9092"
- ],
- "routing": [
- {
- "fromDomain": "perf3gpp",
- "toTopic": "HV_VES_PERF3GPP"
- }
- ]
}
}
import org.onap.dcae.collectors.veshv.model.ServiceContext
import org.onap.dcae.collectors.veshv.utils.ServerHandle
import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import org.onap.dcae.collectors.veshv.utils.neverComplete
import org.onap.dcae.collectors.veshv.utils.registerShutdownHook
import reactor.core.scheduler.Schedulers
import java.util.concurrent.atomic.AtomicReference
HealthState.INSTANCE.changeState(HealthDescription.HEALTHY)
}
.doOnError {
- logger.error(ServiceContext::mdc) { "Failed to acquire configuration ${it.message}" }
- logger.debug(ServiceContext::mdc) { "Detailed stack trace: $it" }
+ logger.error(ServiceContext::mdc) { "Failed to create configuration: ${it.message}" }
+ logger.withDebug(ServiceContext::mdc) { log("Detailed stack trace: ", it) }
HealthState.INSTANCE.changeState(HealthDescription.DYNAMIC_CONFIGURATION_NOT_FOUND)
}
.doOnNext(::startServer)
.doOnError(::logServerStartFailed)
- .neverComplete() // TODO: remove after merging configuration stream with cbs
+ .then()
.block()
}
private fun startServer(config: HvVesConfiguration) {
stopRunningServer()
Logger.setLogLevel(VES_HV_PACKAGE, config.logLevel)
- logger.info { "Using configuration: $config" }
+ logger.debug(ServiceContext::mdc) { "Configuration: $config" }
VesServer.start(config).let {
registerShutdownHook { shutdownGracefully(it) }
-->
<configuration>
<property name="COMPONENT_NAME"
- value="dcae-hv-ves-collector"/>
+ value="dcae-hv-ves-collector"/>
<property name="COMPONENT_SHORT_NAME"
- value="hv-ves"/>
+ value="hv-ves"/>
<property name="LOG_FILENAME" value="${COMPONENT_SHORT_NAME}"/>
<property name="LOG_PATH" value="/var/log/ONAP/${COMPONENT_NAME}"/>
| ${p_mdc}\t
| ${p_thr}%n"/>
- <property name="ONAP_LOG_PATTERN" value="%nopexception${p_log}|${p_tim}|${p_lvl}|${p_msg}|${p_mdc}|${p_exc}|${p_mak}|${p_thr}%n"/>
+ <property name="ONAP_LOG_PATTERN"
+ value="%nopexception${p_log}|${p_tim}|${p_lvl}|${p_msg}|${p_mdc}|${p_exc}|${p_mak}|${p_thr}%n"/>
<property name="ONAP_LOG_PATTERN_FROM_WIKI" value="%nopexception%logger
|%date{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC}
</appender>
<appender name="ROLLING-FILE"
- class="ch.qos.logback.core.rolling.RollingFileAppender">
+ class="ch.qos.logback.core.rolling.RollingFileAppender">
<encoder>
<pattern>${ONAP_LOG_PATTERN}</pattern>
</encoder>
<logger name="io.netty" level="INFO"/>
<logger name="io.netty.util" level="WARN"/>
<logger name="org.apache.kafka" level="INFO"/>
+ <logger name="org.onap.dcaegen2.services.sdk" level="INFO"/>
+ <logger name="org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.CbsClientImpl" level="WARN"/>
- <root level="INFO">
+ <root level="DEBUG">
<appender-ref ref="CONSOLE"/>
<appender-ref ref="ROLLING-FILE"/>
</root>
import arrow.typeclasses.MonadContinuation
import arrow.typeclasses.binding
import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
import java.util.concurrent.atomic.AtomicReference
/**
fun <A, B> Either<A, B>.rightOrThrow(mapper: (A) -> Throwable) = fold({ throw mapper(it) }, ::identity)
+fun <A : Exception, B> Flux<Either<A, B>>.throwOnLeft(): Flux<B> = map { it.rightOrThrow() }
+
fun <A, B> Flux<Either<A, B>>.throwOnLeft(f: (A) -> Exception): Flux<B> = map { it.rightOrThrow(f) }
+fun <A, B> Mono<Either<A, B>>.throwOnLeft(f: (A) -> Exception): Mono<B> = map { it.rightOrThrow(f) }
+
fun <A> AtomicReference<A>.getOption() = Option.fromNullable(get())
fun <A> Option.Companion.fromNullablesChain(firstValue: A?, vararg nextValues: () -> A?): Option<A> =
+++ /dev/null
-/*
- * ============LICENSE_START=======================================================
- * dcaegen2-collectors-veshv
- * ================================================================================
- * Copyright (C) 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=========================================================
- */
-package org.onap.dcae.collectors.veshv.utils
-
-import reactor.core.publisher.Flux
-import reactor.core.publisher.Mono
-
-fun <T> Flux<T>.neverComplete(): Mono<Void> = then(Mono.never<T>()).then()
\ No newline at end of file
Code Review / dcaegen2 / collectors / hv-ves.git / commitdiff