diff options
Diffstat (limited to 'sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageGenerator.kt')
| -rw-r--r-- | sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageGenerator.kt | 20 |
1 files changed, 14 insertions, 6 deletions
diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageGenerator.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageGenerator.kt index 076c06be..5f8638f0 100644 --- a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageGenerator.kt +++ b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageGenerator.kt @@ -2,7 +2,7 @@ * ============LICENSE_START======================================================= * dcaegen2-collectors-veshv * ================================================================================ - * Copyright (C) 2018 NOKIA + * 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. @@ -19,18 +19,26 @@ */ package org.onap.dcae.collectors.veshv.ves.message.generator.api -import org.onap.dcae.collectors.veshv.domain.WireFrameMessage import reactor.core.publisher.Flux +import reactor.core.publisher.Mono /** * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> * @since June 2018 */ -interface MessageGenerator { - fun createMessageFlux(messageParameters: List<MessageParameters>): Flux<WireFrameMessage> +abstract class MessageGenerator<K : MessageParameters, T> { + abstract fun createMessageFlux(parameters: K): Flux<T> - companion object { - const val FIXED_PAYLOAD_SIZE = 100 + protected fun repeatMessage(message: Mono<T>, amount: Long): Flux<T> = when { + amount < 0 -> repeatForever(message) + amount == 0L -> emptyMessageStream() + else -> repeatNTimes(message, amount) } + + private fun repeatForever(message: Mono<T>) = message.repeat() + + private fun emptyMessageStream() = Flux.empty<T>() + + private fun repeatNTimes(message: Mono<T>, amount: Long) = message.repeat(amount - 1) } |