* limitations under the License.
* ============LICENSE_END=========================================================
*/
-package org.onap.dcae.collectors.veshv.tests.component
+package org.onap.dcae.collectors.veshv.simulators.xnf.api
-import org.jetbrains.spek.api.dsl.Pending
-import org.jetbrains.spek.api.dsl.TestContainer
+import org.onap.dcae.collectors.veshv.domain.WireFrame
+import org.onap.dcae.collectors.veshv.simulators.xnf.config.MessageParameters
+import reactor.core.publisher.Flux
-internal fun TestContainer.system(description: String, body: (Sut) -> Unit) {
- test("system $description", body = { body(Sut()) })
-}
-
-internal fun TestContainer.xsystem(description: String, reason: String? = null, body: (Sut) -> Unit = {}) {
- test("system $description", Pending.Yes(reason), body = { body(Sut()) })
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+interface MessageGenerator {
+ fun createMessageFlux(messageParameters: MessageParameters): Flux<WireFrame>
}
* @author Jakub Dudycz <jakub.dudycz@nokia.com>
* @since June 2018
*/
-data class ClientConfiguration(
+internal data class ClientConfiguration(
val vesHost: String,
val vesPort: Int,
val security: SecurityConfiguration,
--- /dev/null
+/*
+ * ============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.simulators.xnf
+
+import org.onap.dcae.collectors.veshv.simulators.xnf.api.MessageGenerator
+import org.onap.dcae.collectors.veshv.simulators.xnf.impl.MessageGeneratorImpl
+import org.onap.dcae.collectors.veshv.simulators.xnf.impl.PayloadGenerator
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+fun createMessageGenerator(): MessageGenerator = MessageGeneratorImpl(PayloadGenerator())
* @author Jakub Dudycz <jakub.dudycz@nokia.com>
* @since June 2018
*/
-class HttpServer(private val vesClient: VesHvClient) {
+internal class HttpServer(private val vesClient: VesHvClient) {
fun start(port: Int = DEFAULT_PORT): IO<RatpackServer> = IO {
RatpackServer.start { server ->
return ctx.request.body
.map { Json.createReader(it.inputStream).readObject() }
.map { extractMessageParameters(it) }
- .map { MessageFactory.INSTANCE.createMessageFlux(it) }
+ .map { MessageGeneratorImpl.INSTANCE.createMessageFlux(it) }
}
private fun sendAcceptedResponse(ctx: Context) {
private fun extractMessageParameters(request: JsonObject): MessageParameters =
try {
- val commonEventHeader = MessageFactory.INSTANCE
+ val commonEventHeader = MessageGeneratorImpl.INSTANCE
.parseCommonHeader(request.getJsonObject("commonEventHeader"))
val messagesAmount = request.getJsonNumber("messagesAmount").longValue()
MessageParameters(commonEventHeader, messagesAmount)
import com.google.protobuf.ByteString
import org.onap.dcae.collectors.veshv.domain.WireFrame
+import org.onap.dcae.collectors.veshv.simulators.xnf.api.MessageGenerator
import org.onap.dcae.collectors.veshv.simulators.xnf.config.MessageParameters
import org.onap.ves.VesEventV5.VesEvent
import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
* @author Jakub Dudycz <jakub.dudycz@nokia.com>
* @since June 2018
*/
-class MessageFactory(private val payloadGenerator: PayloadGenerator) {
+internal class MessageGeneratorImpl(private val payloadGenerator: PayloadGenerator) : MessageGenerator {
- fun createMessageFlux(messageParameters: MessageParameters): Flux<WireFrame> =
+ override fun createMessageFlux(messageParameters: MessageParameters): Flux<WireFrame> =
Mono.fromCallable { createMessage(messageParameters.commonEventHeader) }.let {
if (messageParameters.amount < 0)
it.repeat()
WireFrame(vesMessageBytes(commonHeader))
- private fun vesMessageBytes(commonHeader: CommonEventHeader): ByteArray {
- val msg = VesEvent.newBuilder()
- .setCommonEventHeader(commonHeader)
- .setHvRanMeasFields(PayloadGenerator().generatePayload().toByteString())
- .build()
-
- return msg.toByteArray()
- }
+ private fun vesMessageBytes(commonHeader: CommonEventHeader): ByteArray =
+ VesEvent.newBuilder()
+ .setCommonEventHeader(commonHeader)
+ .setHvRanMeasFields(payloadGenerator.generatePayload().toByteString())
+ .build()
+ .toByteArray()
companion object {
- val INSTANCE = MessageFactory(PayloadGenerator())
+ val INSTANCE = MessageGeneratorImpl(PayloadGenerator())
}
}
import org.onap.ves.HVRanMeasFieldsV5.HVRanMeasFields.HVRanMeasPayload
import org.onap.ves.HVRanMeasFieldsV5.HVRanMeasFields.HVRanMeasPayload.PMObject
import org.onap.ves.HVRanMeasFieldsV5.HVRanMeasFields.HVRanMeasPayload.PMObject.HVRanMeas
-import java.util.Random
+import java.util.*
-class PayloadGenerator {
+internal class PayloadGenerator {
private val randomGenerator = Random()
import org.onap.dcae.collectors.veshv.simulators.xnf.config.ClientConfiguration
import org.onap.dcae.collectors.veshv.utils.logging.Logger
import org.reactivestreams.Publisher
-import reactor.core.publisher.EmitterProcessor
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.ReplayProcessor
* @author Jakub Dudycz <jakub.dudycz@nokia.com>
* @since June 2018
*/
-class VesHvClient(private val configuration: ClientConfiguration) {
+internal class VesHvClient(private val configuration: ClientConfiguration) {
private val client: TcpClient = TcpClient.builder()
.options { opts ->
import arrow.core.Failure
import arrow.core.Success
-import arrow.effects.IO
import org.onap.dcae.collectors.veshv.simulators.xnf.config.ArgBasedClientConfiguration
-import org.onap.dcae.collectors.veshv.simulators.xnf.config.ClientConfiguration
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.HttpServer
-import org.onap.dcae.collectors.veshv.simulators.xnf.impl.MessageFactory
import org.onap.dcae.collectors.veshv.simulators.xnf.impl.VesHvClient
import org.onap.dcae.collectors.veshv.utils.commandline.handleErrorsInMain
import org.onap.dcae.collectors.veshv.utils.logging.Logger
object MessageFactoryTest : Spek({
describe("message factory") {
- val factory = MessageFactory.INSTANCE
+ val factory = MessageGeneratorImpl.INSTANCE
given("only common header") {
it("should return infinite flux") {
.compose(sink::send)
.doOnNext { metrics.notifyMessageSent(it.topic) }
.doOnTerminate { releaseBuffersMemory(wireDecoder) }
+ .onErrorResume(this::handleErrors)
.then()
}
private fun findRoute(msg: VesMessage): Mono<RoutedMessage> = omitWhenNull(msg, router::findDestination)
- private fun releaseBuffersMemory(wireChunkDecoder: WireChunkDecoder) {
- wireChunkDecoder.release()
- }
-
private fun <T, V> omitWhenNull(input: T, mapper: (T) -> Option<V>): Mono<V> =
mapper(input).fold(
{ Mono.empty() },
{ Mono.just(it) })
+ private fun handleErrors(ex: Throwable): Flux<RoutedMessage> {
+ logger.warn("Error while handling message stream: ${ex::class.qualifiedName} (${ex.message})")
+ logger.debug("Detailed stack trace", ex)
+ return Flux.empty()
+ }
+
+ private fun releaseBuffersMemory(wireChunkDecoder: WireChunkDecoder) {
+ wireChunkDecoder.release()
+ }
+
companion object {
- val logger = Logger(VesHvCollector::class)
+ private val logger = Logger(VesHvCollector::class)
}
}
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
* @since May 2018
*/
-internal class WireChunkDecoder(private val decoder: WireFrameDecoder,
- alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT) {
+internal class WireChunkDecoder(
+ private val decoder: WireFrameDecoder,
+ alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT) {
private val streamBuffer = alloc.compositeBuffer()
+
+// TODO: use this implementation and cleanup the rest
+// fun decode(byteBuf: ByteBuf): Flux<WireFrame> = Flux.defer<WireFrame> {
+// if (byteBuf.readableBytes() == 0) {
+// byteBuf.release()
+// Flux.empty()
+// } else {
+// streamBuffer.addComponent(true, byteBuf)
+// Flux.generate { next ->
+// try {
+// val frame = decodeFirstFrameFromBuffer()
+// if (frame == null)
+// next.complete()
+// else
+// next.next(frame)
+// } catch (ex: Exception) {
+// next.error(ex)
+// }
+// }
+// }
+// }.doOnTerminate { streamBuffer.discardReadComponents() }
+//
+//
+// private fun decodeFirstFrameFromBuffer(): WireFrame? =
+// try {
+// decoder.decodeFirst(streamBuffer)
+// } catch (ex: MissingWireFrameBytesException) {
+// logger.trace { "${ex.message} - waiting for more data" }
+// null
+// }
fun decode(byteBuf: ByteBuf): Flux<WireFrame> = StreamBufferEmitter
.createFlux(decoder, streamBuffer, byteBuf)
private val streamBuffer: ByteBuf,
private val sink: FluxSink<WireFrame>,
private val requestedFrameCount: Long) {
+ private var completed = false
fun handleSubscriber() {
- logger.trace { "Decoder buffer capacity before decoding frame: ${streamBuffer.capacity()}" }
+ if (!completed) {
+ logger.trace { "Decoder buffer capacity before decoding frame: ${streamBuffer.capacity()}" }
- try {
- if (requestedFrameCount == Long.MAX_VALUE) {
- logger.trace { "Push based strategy" }
- pushAvailableFrames()
- } else {
- logger.trace { "Pull based strategy - req $requestedFrameCount" }
- pushUpToNumberOfFrames()
+ try {
+ if (requestedFrameCount == Long.MAX_VALUE) {
+ logger.trace { "Push based strategy" }
+ pushAvailableFrames()
+ } else {
+ logger.trace { "Pull based strategy - req $requestedFrameCount" }
+ pushUpToNumberOfFrames()
+ }
+ } catch (ex: Exception) {
+ completed = true
+ sink.error(ex)
}
- } catch (ex: Exception) {
- sink.error(ex)
- }
-
- logger.trace { "Decoder buffer capacity after decoding frame: ${streamBuffer.capacity()}" }
+ logger.trace { "Decoder buffer capacity after decoding frame: ${streamBuffer.capacity()}" }
+ }
}
private fun pushAvailableFrames() {
sink.next(nextFrame)
nextFrame = decodeFirstFrameFromBuffer()
}
+ completed = true
sink.complete()
}
}
}
if (remaining > 0 && nextFrame == null) {
+ completed = true
sink.complete()
}
}
<artifactId>hv-collector-core</artifactId>
<version>${project.parent.version}</version>
</dependency>
+ <dependency>
+ <groupId>${project.parent.groupId}</groupId>
+ <artifactId>hv-collector-client-simulator</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.arrow-kt</groupId>
+ <artifactId>arrow-syntax</artifactId>
+ </dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
--- /dev/null
+/*
+ * ============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.tests.component
+
+import arrow.syntax.function.partially1
+import io.netty.buffer.ByteBuf
+import io.netty.buffer.ByteBufAllocator
+import io.netty.buffer.CompositeByteBuf
+import io.netty.buffer.Unpooled
+import io.netty.buffer.UnpooledByteBufAllocator
+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.it
+import org.onap.dcae.collectors.veshv.domain.WireFrameEncoder
+import org.onap.dcae.collectors.veshv.simulators.xnf.config.MessageParameters
+import org.onap.dcae.collectors.veshv.simulators.xnf.createMessageGenerator
+import org.onap.dcae.collectors.veshv.tests.fakes.CountingSink
+import org.onap.dcae.collectors.veshv.tests.fakes.basicConfiguration
+import reactor.core.publisher.Flux
+import reactor.math.sum
+import java.security.MessageDigest
+import java.time.Duration
+import java.util.*
+import kotlin.system.measureTimeMillis
+
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+object PerformanceSpecification : Spek({
+ describe("VES High Volume Collector performance") {
+ it("should handle multiple clients in reasonable time") {
+ val sink = CountingSink()
+ val sut = Sut(sink)
+ sut.configurationProvider.updateConfiguration(basicConfiguration)
+
+ val numMessages: Long = 300_000
+ val runs = 4
+ val timeout = Duration.ofMinutes((1 + (runs / 2)).toLong())
+
+ val params = MessageParameters(
+ commonEventHeader = vesEvent().commonEventHeader,
+ amount = numMessages)
+ val fluxes = (1.rangeTo(runs)).map {
+ sut.collector.handleConnection(sut.alloc, generateDataStream(sut.alloc, params))
+ }
+ val durationMs = measureTimeMillis {
+ Flux.merge(fluxes).then().block(timeout)
+ }
+
+ val durationSec = durationMs / 1000.0
+ val throughput = sink.count / durationSec
+ println("Processed $runs connections each containing $numMessages msgs.")
+ println("Forwarded ${sink.count / ONE_MILION} Mmsgs in $durationSec seconds, that is $throughput msgs/s")
+ assertThat(sink.count)
+ .describedAs("should send all events")
+ .isEqualTo(runs * numMessages)
+ }
+
+ it("should disconnect on transmission errors") {
+ val sink = CountingSink()
+ val sut = Sut(sink)
+ sut.configurationProvider.updateConfiguration(basicConfiguration)
+
+ val numMessages: Long = 100_000
+ val timeout = Duration.ofSeconds(30)
+
+ val params = MessageParameters(
+ commonEventHeader = vesEvent().commonEventHeader,
+ amount = numMessages)
+
+ val dataStream = generateDataStream(sut.alloc, params)
+ .transform(::dropWhenIndex.partially1 { it % 101 == 0L })
+ sut.collector.handleConnection(sut.alloc, dataStream)
+ .timeout(timeout)
+ .block()
+
+ println("Forwarded ${sink.count} msgs")
+ assertThat(sink.count)
+ .describedAs("should send up to number of events")
+ .isLessThan(numMessages)
+ }
+ }
+
+ describe("test infrastructure") {
+ val digest = MessageDigest.getInstance("MD5")
+
+ fun collectDigest(bb: ByteBuf) {
+ bb.markReaderIndex()
+ while (bb.isReadable) {
+ digest.update(bb.readByte())
+ }
+ bb.resetReaderIndex()
+ }
+
+ fun calculateDigest(arrays: List<ByteArray>): ByteArray {
+ for (array in arrays) {
+ digest.update(array)
+ }
+ return digest.digest()
+ }
+
+ it("should yield same bytes as in the input") {
+ val numberOfBuffers = 10
+ val singleBufferSize = 1000
+ val arrays = (1.rangeTo(numberOfBuffers)).map { randomByteArray(singleBufferSize) }
+ val inputDigest = calculateDigest(arrays)
+
+ val actualTotalSize = Flux.fromIterable(arrays)
+ .map { Unpooled.wrappedBuffer(it) }
+ .transform { simulateRemoteTcp(UnpooledByteBufAllocator.DEFAULT, 4, it) }
+ .doOnNext(::collectDigest)
+ .map {
+ val size = it.readableBytes()
+ it.release()
+ size
+ }
+ .sum()
+ .map(Long::toInt)
+ .block()
+
+ val outputDigest = digest.digest()
+
+ assertThat(actualTotalSize).isEqualTo(numberOfBuffers * singleBufferSize)
+ assertThat(outputDigest).isEqualTo(inputDigest)
+
+ }
+ }
+})
+
+
+private const val ONE_MILION = 1_000_000.0
+
+private val rand = Random()
+private fun randomByteArray(size: Int): ByteArray {
+ val bytes = ByteArray(size)
+ rand.nextBytes(bytes)
+ return bytes
+}
+
+fun dropWhenIndex(predicate: (Long) -> Boolean, stream: Flux<ByteBuf>): Flux<ByteBuf> =
+ stream.index()
+ .filter { predicate(it.t1) }
+ .map { it.t2 }
+
+private fun generateDataStream(alloc: ByteBufAllocator, params: MessageParameters): Flux<ByteBuf> =
+ WireFrameEncoder(alloc).let { encoder ->
+ createMessageGenerator()
+ .createMessageFlux(params)
+ .map(encoder::encode)
+ .transform { simulateRemoteTcp(alloc, 1000, it) }
+ }
+
+private fun simulateRemoteTcp(alloc: ByteBufAllocator, maxSize: Int, byteBuffers: Flux<ByteBuf>) =
+ byteBuffers
+ .bufferTimeout(maxSize, Duration.ofMillis(250))
+ .map { joinBuffers(alloc, it) }
+ .concatMap { randomlySplitTcpFrames(it) }
+
+private fun joinBuffers(alloc: ByteBufAllocator, it: List<ByteBuf>?) =
+ alloc.compositeBuffer().addComponents(true, it)
+
+private fun randomlySplitTcpFrames(bb: CompositeByteBuf): Flux<ByteBuf> {
+ val targetFrameSize = Math.max(4, (bb.readableBytes() * Math.random()).toInt())
+ return Flux.create<ByteBuf> { sink ->
+ while (bb.isReadable) {
+ val frameSize = Math.min(targetFrameSize, bb.readableBytes())
+ sink.next(bb.retainedSlice(bb.readerIndex(), frameSize))
+ bb.readerIndex(bb.readerIndex() + frameSize)
+ }
+ bb.release()
+ sink.complete()
+ }
+}
import io.netty.buffer.ByteBuf
import io.netty.buffer.UnpooledByteBufAllocator
import org.onap.dcae.collectors.veshv.boundary.Collector
+import org.onap.dcae.collectors.veshv.boundary.Sink
import org.onap.dcae.collectors.veshv.boundary.SinkProvider
-import org.onap.dcae.collectors.veshv.model.RoutedMessage
import org.onap.dcae.collectors.veshv.factory.CollectorFactory
+import org.onap.dcae.collectors.veshv.model.RoutedMessage
import org.onap.dcae.collectors.veshv.tests.fakes.FakeConfigurationProvider
import org.onap.dcae.collectors.veshv.tests.fakes.FakeMetrics
-import org.onap.dcae.collectors.veshv.tests.fakes.FakeSink
+import org.onap.dcae.collectors.veshv.tests.fakes.StoringSink
import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import reactor.core.Exceptions
import reactor.core.publisher.Flux
import java.time.Duration
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
* @since May 2018
*/
-internal class Sut {
+class Sut(sink: Sink = StoringSink()) {
val configurationProvider = FakeConfigurationProvider()
- val sink = FakeSink()
+
val alloc = UnpooledByteBufAllocator.DEFAULT
val metrics = FakeMetrics()
private val collectorFactory = CollectorFactory(configurationProvider, SinkProvider.just(sink), metrics)
val collector: Collector
get() = collectorProvider()
+}
- fun handleConnection(vararg packets: ByteBuf): List<RoutedMessage> {
- collector.handleConnection(alloc, Flux.fromArray(packets)).block(Duration.ofSeconds(10))
- return sink.sentMessages
- }
-
- fun handleConnectionReturningError(vararg packets: ByteBuf): Pair<List<RoutedMessage>, Exception?> =
- try {
- collector.handleConnection(alloc, Flux.fromArray(packets)).block(Duration.ofSeconds(10))
- Pair(sink.sentMessages, null)
- } catch (ex: Exception) {
- Pair(sink.sentMessages, ex)
- }
-
- companion object {
- val logger = Logger(Sut::class)
- }
+fun Sut.handleConnection(sink: StoringSink, vararg packets: ByteBuf): List<RoutedMessage> {
+ collector.handleConnection(alloc, Flux.fromArray(packets)).block(Duration.ofSeconds(10))
+ return sink.sentMessages
}
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.it
import org.onap.dcae.collectors.veshv.domain.exceptions.InvalidWireFrameMarkerException
import org.onap.dcae.collectors.veshv.tests.fakes.HVRANMEAS_TOPIC
+import org.onap.dcae.collectors.veshv.tests.fakes.StoringSink
import org.onap.dcae.collectors.veshv.tests.fakes.basicConfiguration
import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
*/
object VesHvSpecification : Spek({
describe("VES High Volume Collector") {
- system("should handle multiple HV RAN events") { sut ->
+ it("should handle multiple HV RAN events") {
+ val sink = StoringSink()
+ val sut = Sut(sink)
sut.configurationProvider.updateConfiguration(basicConfiguration)
- val messages = sut.handleConnection(vesMessage(Domain.HVRANMEAS), vesMessage(Domain.HVRANMEAS))
+ val messages = sut.handleConnection(sink, vesMessage(Domain.HVRANMEAS), vesMessage(Domain.HVRANMEAS))
assertThat(messages)
.describedAs("should send all events")
describe("Memory management") {
- system("should release memory for each handled and dropped message") { sut ->
+ it("should release memory for each handled and dropped message") {
+ val sink = StoringSink()
+ val sut = Sut(sink)
sut.configurationProvider.updateConfiguration(basicConfiguration)
val validMessage = vesMessage(Domain.HVRANMEAS)
val msgWithInvalidDomain = vesMessage(Domain.OTHER)
val msgWithInvalidFrame = invalidWireFrame()
val expectedRefCnt = 0
- val (handledEvents, exception) = sut.handleConnectionReturningError(
- validMessage, msgWithInvalidDomain, msgWithInvalidFrame)
+ val handledEvents = sut.handleConnection(sink, validMessage, msgWithInvalidDomain, msgWithInvalidFrame)
assertThat(handledEvents).hasSize(1)
- assertThat(exception).isNull()
assertThat(validMessage.refCnt())
.describedAs("handled message should be released")
}
- system("should release memory for each message with invalid payload") { sut ->
+ it("should release memory for each message with invalid payload") {
+ val sink = StoringSink()
+ val sut = Sut(sink)
sut.configurationProvider.updateConfiguration(basicConfiguration)
val validMessage = vesMessage(Domain.HVRANMEAS)
val msgWithInvalidPayload = invalidVesMessage()
val expectedRefCnt = 0
- val (handledEvents, exception) = sut.handleConnectionReturningError(
- validMessage, msgWithInvalidPayload)
+ val handledEvents = sut.handleConnection(sink, validMessage, msgWithInvalidPayload)
assertThat(handledEvents).hasSize(1)
- assertThat(exception?.cause).isInstanceOf(InvalidProtocolBufferException::class.java)
assertThat(validMessage.refCnt())
.describedAs("handled message should be released")
}
- system("should release memory for each message with garbage frame") { sut ->
+ it("should release memory for each message with garbage frame") {
+ val sink = StoringSink()
+ val sut = Sut(sink)
sut.configurationProvider.updateConfiguration(basicConfiguration)
val validMessage = vesMessage(Domain.HVRANMEAS)
val msgWithGarbageFrame = garbageFrame()
val expectedRefCnt = 0
- val (handledEvents, exception) = sut.handleConnectionReturningError(
- validMessage, msgWithGarbageFrame)
+ val handledEvents = sut.handleConnection(sink, validMessage, msgWithGarbageFrame)
assertThat(handledEvents).hasSize(1)
- assertThat(exception?.cause)
- .isInstanceOf(InvalidWireFrameMarkerException::class.java)
assertThat(validMessage.refCnt())
.describedAs("handled message should be released")
}
describe("message routing") {
- system("should direct message to a topic by means of routing configuration") { sut ->
+ it("should direct message to a topic by means of routing configuration") {
+ val sink = StoringSink()
+ val sut = Sut(sink)
sut.configurationProvider.updateConfiguration(basicConfiguration)
- val messages = sut.handleConnection(vesMessage(Domain.HVRANMEAS))
+ val messages = sut.handleConnection(sink, vesMessage(Domain.HVRANMEAS))
assertThat(messages).describedAs("number of routed messages").hasSize(1)
val msg = messages[0]
assertThat(msg.partition).describedAs("routed message partition").isEqualTo(0)
}
- system("should drop message if route was not found") { sut ->
+ it("should drop message if route was not found") {
+ val sink = StoringSink()
+ val sut = Sut(sink)
sut.configurationProvider.updateConfiguration(basicConfiguration)
- val messages = sut.handleConnection(
+ val messages = sut.handleConnection(sink,
vesMessage(Domain.OTHER, "first"),
vesMessage(Domain.HVRANMEAS, "second"),
vesMessage(Domain.HEARTBEAT, "third"))
import org.onap.ves.VesEventV5.VesEvent
import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
-import java.util.UUID
+import java.util.*
val allocator: ByteBufAllocator = PooledByteBufAllocator.DEFAULT
writeByte(0x01) // content type = GPB
}
-fun vesEvent(domain: Domain = Domain.OTHER, id: String = UUID.randomUUID().toString()) =
+fun vesEvent(domain: Domain = Domain.HVRANMEAS, id: String = UUID.randomUUID().toString()) =
VesEvent.newBuilder()
.setCommonEventHeader(
CommonEventHeader.getDefaultInstance().toBuilder()
import org.onap.dcae.collectors.veshv.boundary.Sink
import org.onap.dcae.collectors.veshv.model.RoutedMessage
-import org.onap.dcae.collectors.veshv.model.VesMessage
import reactor.core.publisher.Flux
import java.util.*
import java.util.concurrent.ConcurrentLinkedDeque
+import java.util.concurrent.atomic.AtomicLong
/**
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
* @since May 2018
*/
-class FakeSink : Sink {
+class StoringSink : Sink {
private val sent: Deque<RoutedMessage> = ConcurrentLinkedDeque()
val sentMessages: List<RoutedMessage>
return messages.doOnNext(sent::addLast)
}
}
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+class CountingSink : Sink {
+ private val atomicCount = AtomicLong(0)
+
+ val count: Long
+ get() = atomicCount.get()
+
+ override fun send(messages: Flux<RoutedMessage>): Flux<RoutedMessage> {
+ return messages.doOnNext {
+ atomicCount.incrementAndGet()
+ }
+ }
+}
</rollingPolicy>
</appender>
- <logger name="org.onap.dcae.collectors.veshv" level="TRACE"/>
+ <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
<root level="INFO">
<appender-ref ref="CONSOLE"/>
<appender-ref ref="ROLLING-FILE"/>
</root>
-</configuration>
\ No newline at end of file
+</configuration>
--- /dev/null
+{
+ "commonEventHeader": {
+ "version": "sample-version",
+ "domain": 10,
+ "sequence": 1,
+ "priority": 1,
+ "eventId": "sample-event-id",
+ "eventName": "sample-event-name",
+ "eventType": "sample-event-type",
+ "startEpochMicrosec": 120034455,
+ "lastEpochMicrosec": 120034455,
+ "nfNamingCode": "sample-nf-naming-code",
+ "nfcNamingCode": "sample-nfc-naming-code",
+ "reportingEntityId": "sample-reporting-entity-id",
+ "reportingEntityName": "sample-reporting-entity-name",
+ "sourceId": "sample-source-id",
+ "sourceName": "sample-source-name"
+ },
+ "messagesAmount": 1000000
+}
Code Review / dcaegen2 / collectors / hv-ves.git / commitdiff