diff options
Diffstat (limited to 'hv-collector-core')
18 files changed, 188 insertions, 100 deletions
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/CollectorFactory.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/CollectorFactory.kt index 913d8f50..73f4d09d 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/CollectorFactory.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/CollectorFactory.kt @@ -23,12 +23,13 @@ import org.onap.dcae.collectors.veshv.boundary.Collector import org.onap.dcae.collectors.veshv.boundary.CollectorProvider import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider import org.onap.dcae.collectors.veshv.boundary.SinkProvider -import org.onap.dcae.collectors.veshv.model.CollectorConfiguration +import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder import org.onap.dcae.collectors.veshv.impl.MessageValidator import org.onap.dcae.collectors.veshv.impl.Router import org.onap.dcae.collectors.veshv.impl.VesDecoder import org.onap.dcae.collectors.veshv.impl.VesHvCollector -import org.onap.dcae.collectors.veshv.impl.wire.WireDecoder +import org.onap.dcae.collectors.veshv.impl.wire.WireChunkDecoder +import org.onap.dcae.collectors.veshv.model.CollectorConfiguration import reactor.core.publisher.Flux import java.util.concurrent.atomic.AtomicReference @@ -36,7 +37,8 @@ import java.util.concurrent.atomic.AtomicReference * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> * @since May 2018 */ -class CollectorFactory(val configuration: ConfigurationProvider, val sinkProvider: SinkProvider) { +class CollectorFactory(val configuration: ConfigurationProvider, + private val sinkProvider: SinkProvider) { fun createVesHvCollectorProvider(): CollectorProvider { val collector: AtomicReference<Collector> = AtomicReference() createVesHvCollector().subscribe(collector::set) @@ -48,7 +50,7 @@ class CollectorFactory(val configuration: ConfigurationProvider, val sinkProvide private fun createVesHvCollector(config: CollectorConfiguration): Collector { return VesHvCollector( - { WireDecoder(it) }, + { alloc -> WireChunkDecoder(WireFrameDecoder(), alloc) }, VesDecoder(), MessageValidator(), Router(config.routing), diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoder.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoder.kt index 60e7d70a..591a48b7 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoder.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoder.kt @@ -19,9 +19,8 @@ */ package org.onap.dcae.collectors.veshv.impl -import io.netty.buffer.ByteBuf +import org.onap.dcae.collectors.veshv.domain.ByteData import org.onap.dcae.collectors.veshv.model.VesMessage -import org.onap.dcae.collectors.veshv.utils.logging.Logger import org.onap.ves.VesEventV5.VesEvent /** @@ -30,8 +29,8 @@ import org.onap.ves.VesEventV5.VesEvent */ internal class VesDecoder { - fun decode(bb: ByteBuf): VesMessage { - val decodedHeader = VesEvent.parseFrom(bb.nioBuffer()).commonEventHeader - return VesMessage(decodedHeader, bb) + fun decode(bytes: ByteData): VesMessage { + val decodedHeader = VesEvent.parseFrom(bytes.unsafeAsArray()).commonEventHeader + return VesMessage(decodedHeader, bytes) } } diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesHvCollector.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesHvCollector.kt index ac11b3e8..965943f6 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesHvCollector.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesHvCollector.kt @@ -24,57 +24,42 @@ import io.netty.buffer.ByteBufAllocator import org.onap.dcae.collectors.veshv.boundary.Collector import org.onap.dcae.collectors.veshv.boundary.Sink import org.onap.dcae.collectors.veshv.domain.WireFrame -import org.onap.dcae.collectors.veshv.impl.wire.WireDecoder +import org.onap.dcae.collectors.veshv.impl.wire.WireChunkDecoder import org.onap.dcae.collectors.veshv.model.RoutedMessage import org.onap.dcae.collectors.veshv.model.VesMessage import org.onap.dcae.collectors.veshv.utils.logging.Logger import reactor.core.publisher.Flux import reactor.core.publisher.Mono -import java.util.concurrent.atomic.AtomicInteger /** * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> * @since May 2018 */ internal class VesHvCollector( - private val wireDecoderSupplier: (ByteBufAllocator) -> WireDecoder, + private val wireChunkDecoderSupplier: (ByteBufAllocator) -> WireChunkDecoder, private val protobufDecoder: VesDecoder, private val validator: MessageValidator, private val router: Router, private val sink: Sink) : Collector { override fun handleConnection(alloc: ByteBufAllocator, dataStream: Flux<ByteBuf>): Mono<Void> = - wireDecoderSupplier(alloc).let { wireDecoder -> + wireChunkDecoderSupplier(alloc).let { wireDecoder -> dataStream .concatMap(wireDecoder::decode) .filter(WireFrame::isValid) .map(WireFrame::payload) .map(protobufDecoder::decode) - .filter(this::validate) + .filter(validator::isValid) .flatMap(this::findRoute) .compose(sink::send) - .doOnNext(this::releaseMemory) .doOnTerminate { releaseBuffersMemory(wireDecoder) } .then() } - private fun validate(msg: VesMessage): Boolean { - val valid = validator.isValid(msg) - if (!valid) { - msg.rawMessage.release() - } - return valid - } - private fun findRoute(msg: VesMessage): Mono<RoutedMessage> = omitWhenNull(msg, router::findDestination) - private fun releaseMemory(msg: VesMessage) { - logger.trace { "Releasing memory from ${msg.rawMessage}" } - msg.rawMessage.release() - } - - private fun releaseBuffersMemory(wireDecoder: WireDecoder) { - wireDecoder.release() + private fun releaseBuffersMemory(wireChunkDecoder: WireChunkDecoder) { + wireChunkDecoder.release() } private fun <T, V> omitWhenNull(input: T, mapper: (T) -> V?): Mono<V> = Mono.justOrEmpty(mapper(input)) diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/AdapterFactory.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/AdapterFactory.kt index 8e6db2af..358be108 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/AdapterFactory.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/AdapterFactory.kt @@ -21,6 +21,7 @@ package org.onap.dcae.collectors.veshv.impl.adapters import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider import org.onap.dcae.collectors.veshv.boundary.SinkProvider +import org.onap.dcae.collectors.veshv.impl.adapters.kafka.KafkaSinkProvider import org.onap.dcae.collectors.veshv.model.CollectorConfiguration import reactor.core.publisher.Flux import reactor.ipc.netty.http.client.HttpClient diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/LoggingSinkProvider.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/LoggingSinkProvider.kt index 62b6d1aa..b943e4e5 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/LoggingSinkProvider.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/LoggingSinkProvider.kt @@ -47,7 +47,7 @@ internal class LoggingSinkProvider : SinkProvider { private fun logMessage(msg: RoutedMessage) { val msgs = totalMessages.addAndGet(1) - val bytes = totalBytes.addAndGet(msg.message.rawMessage.readableBytes().toLong()) + val bytes = totalBytes.addAndGet(msg.message.rawMessage.size().toLong()) val logMessageSupplier = { "Message routed to ${msg.topic}. Total = $msgs ($bytes B)" } if (msgs % INFO_LOGGING_FREQ == 0L) logger.info(logMessageSupplier) diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/KafkaSink.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSink.kt index db7845c7..6142fa3c 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/KafkaSink.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSink.kt @@ -17,7 +17,7 @@ * limitations under the License. * ============LICENSE_END========================================================= */ -package org.onap.dcae.collectors.veshv.impl.adapters +package org.onap.dcae.collectors.veshv.impl.adapters.kafka import org.onap.dcae.collectors.veshv.boundary.Sink import org.onap.dcae.collectors.veshv.model.RoutedMessage @@ -28,13 +28,12 @@ import reactor.core.publisher.Flux import reactor.kafka.sender.KafkaSender import reactor.kafka.sender.SenderRecord import reactor.kafka.sender.SenderResult -import java.nio.ByteBuffer /** * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> * @since May 2018 */ -internal class KafkaSink(private val sender: KafkaSender<CommonEventHeader, ByteBuffer>) : Sink { +internal class KafkaSink(private val sender: KafkaSender<CommonEventHeader, VesMessage>) : Sink { override fun send(messages: Flux<RoutedMessage>): Flux<VesMessage> { val records = messages.map(this::vesToKafkaRecord) @@ -44,13 +43,13 @@ internal class KafkaSink(private val sender: KafkaSender<CommonEventHeader, Byte .map { it.correlationMetadata() } } - private fun vesToKafkaRecord(msg: RoutedMessage): SenderRecord<CommonEventHeader, ByteBuffer, VesMessage> { + private fun vesToKafkaRecord(msg: RoutedMessage): SenderRecord<CommonEventHeader, VesMessage, VesMessage> { return SenderRecord.create( msg.topic, msg.partition, System.currentTimeMillis(), msg.message.header, - msg.message.rawMessage.nioBuffer(), + msg.message, msg.message) } diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/KafkaSinkProvider.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkProvider.kt index 82452e1e..a00a02d2 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/KafkaSinkProvider.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkProvider.kt @@ -17,18 +17,16 @@ * limitations under the License. * ============LICENSE_END========================================================= */ -package org.onap.dcae.collectors.veshv.impl.adapters +package org.onap.dcae.collectors.veshv.impl.adapters.kafka import org.apache.kafka.clients.producer.ProducerConfig -import org.apache.kafka.common.serialization.ByteBufferSerializer -import org.apache.kafka.common.serialization.StringSerializer import org.onap.dcae.collectors.veshv.boundary.Sink import org.onap.dcae.collectors.veshv.boundary.SinkProvider import org.onap.dcae.collectors.veshv.model.CollectorConfiguration -import org.onap.ves.VesEventV5 +import org.onap.dcae.collectors.veshv.model.VesMessage +import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader import reactor.kafka.sender.KafkaSender import reactor.kafka.sender.SenderOptions -import java.nio.ByteBuffer /** * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> @@ -40,9 +38,8 @@ internal class KafkaSinkProvider : SinkProvider { } private fun constructSenderOptions(config: CollectorConfiguration) = - SenderOptions.create<VesEventV5.VesEvent.CommonEventHeader, ByteBuffer>() + SenderOptions.create<CommonEventHeader, VesMessage>() .producerProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.kafkaBootstrapServers) - .producerProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java) - .producerProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteBufferSerializer::class.java) - + .producerProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ProtobufSerializer::class.java) + .producerProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, VesMessageSerializer::class.java) } diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/ProtobufSerializer.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/ProtobufSerializer.kt new file mode 100644 index 00000000..9753d9e5 --- /dev/null +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/ProtobufSerializer.kt @@ -0,0 +1,40 @@ +/* + * ============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.impl.adapters.kafka + +import com.google.protobuf.MessageLite +import org.apache.kafka.common.serialization.Serializer + +/** + * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> + * @since June 2018 + */ +class ProtobufSerializer :Serializer<MessageLite> { + override fun configure(configs: MutableMap<String, *>?, isKey: Boolean) { + // no configuration + } + + override fun serialize(topic: String?, data: MessageLite?): ByteArray? = + data?.toByteArray() + + override fun close() { + // cleanup not needed + } +} diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/VesMessageSerializer.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/VesMessageSerializer.kt new file mode 100644 index 00000000..7a6ac7c8 --- /dev/null +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/VesMessageSerializer.kt @@ -0,0 +1,37 @@ +/* + * ============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.impl.adapters.kafka + +import org.apache.kafka.common.serialization.Serializer +import org.onap.dcae.collectors.veshv.model.VesMessage + +/** + * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> + * @since June 2018 + */ +class VesMessageSerializer : Serializer<VesMessage> { + override fun configure(configs: MutableMap<String, *>?, isKey: Boolean) { + } + + override fun serialize(topic: String?, msg: VesMessage?): ByteArray? = msg?.rawMessage?.unsafeAsArray() + + override fun close() { + } +} diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/StreamBufferEmitter.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/StreamBufferEmitter.kt index e4dd7cf6..34a8b928 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/StreamBufferEmitter.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/StreamBufferEmitter.kt @@ -22,6 +22,7 @@ package org.onap.dcae.collectors.veshv.impl.wire import io.netty.buffer.ByteBuf import io.netty.buffer.CompositeByteBuf import org.onap.dcae.collectors.veshv.domain.WireFrame +import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder import org.onap.dcae.collectors.veshv.utils.logging.Logger import reactor.core.publisher.Flux import reactor.core.publisher.FluxSink @@ -33,6 +34,7 @@ import java.util.function.Consumer * @since May 2018 */ internal class StreamBufferEmitter( + private val decoder: WireFrameDecoder, private val streamBuffer: CompositeByteBuf, private val newFrame: ByteBuf) : Consumer<FluxSink<WireFrame>> { @@ -58,15 +60,15 @@ internal class StreamBufferEmitter( streamBuffer.discardReadComponents() } sink.onRequest { requestedFrameCount -> - WireFrameSink(streamBuffer, sink, requestedFrameCount).handleSubscriber() + WireFrameSink(decoder, streamBuffer, sink, requestedFrameCount).handleSubscriber() } } } } companion object { - fun createFlux(streamBuffer: CompositeByteBuf, newFrame: ByteBuf): Flux<WireFrame> = - Flux.create(StreamBufferEmitter(streamBuffer, newFrame)) + fun createFlux(decoder: WireFrameDecoder, streamBuffer: CompositeByteBuf, newFrame: ByteBuf): Flux<WireFrame> = + Flux.create(StreamBufferEmitter(decoder, streamBuffer, newFrame)) private const val INCREASE_WRITER_INDEX = true private val logger = Logger(StreamBufferEmitter::class) diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireDecoder.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoder.kt index b701aaf2..580d36c5 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireDecoder.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoder.kt @@ -22,6 +22,7 @@ package org.onap.dcae.collectors.veshv.impl.wire import io.netty.buffer.ByteBuf import io.netty.buffer.ByteBufAllocator import org.onap.dcae.collectors.veshv.domain.WireFrame +import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder import org.onap.dcae.collectors.veshv.impl.VesHvCollector import org.onap.dcae.collectors.veshv.utils.logging.Logger import reactor.core.publisher.Flux @@ -30,10 +31,10 @@ import reactor.core.publisher.Flux * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> * @since May 2018 */ -internal class WireDecoder(alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT) { +internal class WireChunkDecoder(private val decoder: WireFrameDecoder, alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT) { private val streamBuffer = alloc.compositeBuffer() - fun decode(byteBuf: ByteBuf): Flux<WireFrame> = StreamBufferEmitter.createFlux(streamBuffer, byteBuf) + fun decode(byteBuf: ByteBuf): Flux<WireFrame> = StreamBufferEmitter.createFlux(decoder, streamBuffer, byteBuf) .doOnSubscribe { logIncomingMessage(byteBuf) } .doOnNext(this::logDecodedWireMessage) @@ -41,7 +42,6 @@ internal class WireDecoder(alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT) { streamBuffer.release() } - private fun logIncomingMessage(wire: ByteBuf) { logger.trace { "Got message with total size of ${wire.readableBytes()} B" } } diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireFrameSink.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireFrameSink.kt index bc9c8389..a576dc65 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireFrameSink.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireFrameSink.kt @@ -21,6 +21,7 @@ package org.onap.dcae.collectors.veshv.impl.wire import io.netty.buffer.ByteBuf import org.onap.dcae.collectors.veshv.domain.WireFrame +import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder import org.onap.dcae.collectors.veshv.domain.exceptions.MissingWireFrameBytesException import org.onap.dcae.collectors.veshv.utils.logging.Logger import reactor.core.publisher.FluxSink @@ -30,6 +31,7 @@ import reactor.core.publisher.FluxSink * @since May 2018 */ internal class WireFrameSink( + private val decoder: WireFrameDecoder, private val streamBuffer: ByteBuf, private val sink: FluxSink<WireFrame>, private val requestedFrameCount: Long) { @@ -80,7 +82,7 @@ internal class WireFrameSink( private fun decodeFirstFrameFromBuffer(): WireFrame? = try { - WireFrame.decodeFirst(streamBuffer) + decoder.decodeFirst(streamBuffer) } catch (ex: MissingWireFrameBytesException) { logger.debug { "${ex.message} - waiting for more data" } null diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/VesMessage.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/VesMessage.kt index 38256896..03c53e10 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/VesMessage.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/VesMessage.kt @@ -19,11 +19,11 @@ */ package org.onap.dcae.collectors.veshv.model -import io.netty.buffer.ByteBuf +import org.onap.dcae.collectors.veshv.domain.ByteData import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader /** * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> * @since May 2018 */ -data class VesMessage(val header: CommonEventHeader, val rawMessage: ByteBuf) +data class VesMessage(val header: CommonEventHeader, val rawMessage: ByteData) diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/routing.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/routing.kt index 10e79156..bc030587 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/routing.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/routing.kt @@ -73,7 +73,7 @@ class RouteBuilder { this.targetTopic = targetTopic } - fun withFixedPartitioning(num: Int = 1) { + fun withFixedPartitioning(num: Int = 0) { partitioning = { _ -> num } } diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/MessageValidatorTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/MessageValidatorTest.kt index df2840b9..017187a4 100644 --- a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/MessageValidatorTest.kt +++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/MessageValidatorTest.kt @@ -20,27 +20,29 @@ package org.onap.dcae.collectors.veshv.impl import com.google.protobuf.ByteString -import io.netty.buffer.ByteBuf -import io.netty.buffer.Unpooled -import io.netty.buffer.Unpooled.wrappedBuffer +import org.assertj.core.api.Assertions.assertThat import org.jetbrains.spek.api.Spek import org.jetbrains.spek.api.dsl.given import org.jetbrains.spek.api.dsl.it import org.jetbrains.spek.api.dsl.on +import org.onap.dcae.collectors.veshv.domain.ByteData +import org.onap.dcae.collectors.veshv.domain.toByteData import org.onap.dcae.collectors.veshv.model.VesMessage -import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader import org.onap.ves.VesEventV5.VesEvent -import org.assertj.core.api.Assertions.assertThat -import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.* +import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader +import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain +import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Priority +import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.getDefaultInstance +import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.newBuilder internal object MessageValidatorTest : Spek({ - fun vesMessageBytes(commonHeader: CommonEventHeader): ByteBuf { + fun vesMessageBytes(commonHeader: CommonEventHeader): ByteData { val msg = VesEvent.newBuilder() .setCommonEventHeader(commonHeader) .setHvRanMeasFields(ByteString.copyFromUtf8("high volume data")) .build() - return wrappedBuffer(msg.toByteArray()) + return msg.toByteData() } given("Message validator") { @@ -79,7 +81,7 @@ internal object MessageValidatorTest : Spek({ } on("ves hv message bytes") { - val vesMessage = VesMessage(getDefaultInstance(), Unpooled.EMPTY_BUFFER) + val vesMessage = VesMessage(getDefaultInstance(), ByteData.EMPTY) it("should not accept message with default header") { assertThat(cut.isValid(vesMessage)).describedAs("message validation result").isFalse() } @@ -97,7 +99,7 @@ internal object MessageValidatorTest : Spek({ .setCommonEventHeader(commonHeader) .setHvRanMeasFields(ByteString.copyFromUtf8("high volume data !!!")) .build() - val rawMessageBytes = wrappedBuffer(msg.toByteArray()) + val rawMessageBytes = msg.toByteData() it("should not accept not fully initialized message header ") { val vesMessage = VesMessage(commonHeader, rawMessageBytes) diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/RouterTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/RouterTest.kt index 3812db58..c852f5f4 100644 --- a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/RouterTest.kt +++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/RouterTest.kt @@ -1,11 +1,30 @@ +/* + * ============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.impl -import io.netty.buffer.Unpooled import org.assertj.core.api.Assertions.assertThat import org.jetbrains.spek.api.Spek import org.jetbrains.spek.api.dsl.given import org.jetbrains.spek.api.dsl.it import org.jetbrains.spek.api.dsl.on +import org.onap.dcae.collectors.veshv.domain.ByteData import org.onap.dcae.collectors.veshv.model.VesMessage import org.onap.dcae.collectors.veshv.model.routing import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader @@ -34,7 +53,7 @@ object RouterTest : Spek({ val cut = Router(config) on("message with existing route (rtpm)") { - val message = VesMessage(vesCommonHeaderWithDomain(Domain.HVRANMEAS), Unpooled.EMPTY_BUFFER) + val message = VesMessage(vesCommonHeaderWithDomain(Domain.HVRANMEAS), ByteData.EMPTY) val result = cut.findDestination(message) it("should have route available") { @@ -55,7 +74,7 @@ object RouterTest : Spek({ } on("message with existing route (trace)") { - val message = VesMessage(vesCommonHeaderWithDomain(Domain.SYSLOG), Unpooled.EMPTY_BUFFER) + val message = VesMessage(vesCommonHeaderWithDomain(Domain.SYSLOG), ByteData.EMPTY) val result = cut.findDestination(message) it("should have route available") { @@ -63,7 +82,7 @@ object RouterTest : Spek({ } it("should be routed to proper partition") { - assertThat(result?.partition).isEqualTo(1) + assertThat(result?.partition).isEqualTo(0) } it("should be routed to proper topic") { @@ -76,7 +95,7 @@ object RouterTest : Spek({ } on("message with unknown route") { - val message = VesMessage(vesCommonHeaderWithDomain(Domain.HEARTBEAT), Unpooled.EMPTY_BUFFER) + val message = VesMessage(vesCommonHeaderWithDomain(Domain.HEARTBEAT), ByteData.EMPTY) val result = cut.findDestination(message) it("should not have route available") { diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoderTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoderTest.kt index 263ad441..90b34b1c 100644 --- a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoderTest.kt +++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoderTest.kt @@ -21,13 +21,14 @@ package org.onap.dcae.collectors.veshv.impl import com.google.protobuf.ByteString import com.google.protobuf.InvalidProtocolBufferException -import io.netty.buffer.Unpooled.wrappedBuffer import org.assertj.core.api.Assertions.assertThat import org.assertj.core.api.Assertions.assertThatExceptionOfType import org.jetbrains.spek.api.Spek import org.jetbrains.spek.api.dsl.given import org.jetbrains.spek.api.dsl.it import org.jetbrains.spek.api.dsl.on +import org.onap.dcae.collectors.veshv.domain.ByteData +import org.onap.dcae.collectors.veshv.domain.toByteData import org.onap.dcae.collectors.veshv.model.VesMessage import org.onap.ves.VesEventV5.VesEvent import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader @@ -45,7 +46,7 @@ internal object VesDecoderTest : Spek({ .setCommonEventHeader(commonHeader) .setHvRanMeasFields(ByteString.copyFromUtf8("highvolume measurements")) .build() - val rawMessageBytes = wrappedBuffer(msg.toByteArray()) + val rawMessageBytes = msg.toByteData() it("should decode only header and pass it on along with raw message") { @@ -60,7 +61,7 @@ internal object VesDecoderTest : Spek({ } on("invalid ves hv message bytes") { - val rawMessageBytes = wrappedBuffer("ala ma kota".toByteArray(Charset.defaultCharset())) + val rawMessageBytes = ByteData("ala ma kota".toByteArray(Charset.defaultCharset())) it("should throw error") { assertThatExceptionOfType(InvalidProtocolBufferException::class.java) diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireDecoderTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoderTest.kt index 0a10aa1f..1ddcc3dc 100644 --- a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireDecoderTest.kt +++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoderTest.kt @@ -28,6 +28,8 @@ 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.domain.WireFrame +import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder +import org.onap.dcae.collectors.veshv.domain.WireFrameEncoder import org.onap.dcae.collectors.veshv.domain.exceptions.InvalidWireFrameMarkerException import reactor.test.test @@ -35,13 +37,17 @@ import reactor.test.test * @author Piotr Jaszczyk <piotr.jaszczyk></piotr.jaszczyk>@nokia.com> * @since May 2018 */ -internal object WireDecoderTest : Spek({ +internal object WireChunkDecoderTest : Spek({ val alloc = UnpooledByteBufAllocator.DEFAULT val samplePayload = "konstantynopolitanczykowianeczka".toByteArray() val anotherPayload = "ala ma kota a kot ma ale".toByteArray() - fun WireDecoder.decode(frame: WireFrame) = decode(frame.encode(alloc)) + val encoder = WireFrameEncoder(alloc) + + fun WireChunkDecoder.decode(frame: WireFrame) = decode(encoder.encode(frame)) + fun createInstance() = WireChunkDecoder(WireFrameDecoder(), alloc) + fun verifyMemoryReleased(vararg byteBuffers: ByteBuf) { for (bb in byteBuffers) { assertThat(bb.refCnt()) @@ -63,7 +69,7 @@ internal object WireDecoderTest : Spek({ val input = Unpooled.EMPTY_BUFFER it("should yield empty result") { - WireDecoder().decode(input).test().verifyComplete() + createInstance().decode(input).test().verifyComplete() } } @@ -71,7 +77,7 @@ internal object WireDecoderTest : Spek({ val input = Unpooled.wrappedBuffer(byteArrayOf(0x00)).readerIndex(1) it("should yield empty result") { - WireDecoder().decode(input).test().verifyComplete() + createInstance().decode(input).test().verifyComplete() } it("should release memory") { @@ -83,7 +89,7 @@ internal object WireDecoderTest : Spek({ val input = Unpooled.wrappedBuffer(samplePayload) it("should yield error") { - WireDecoder().decode(input).test() + createInstance().decode(input).test() .verifyError(InvalidWireFrameMarkerException::class.java) } @@ -93,10 +99,10 @@ internal object WireDecoderTest : Spek({ } given("valid input") { - val input = WireFrame(Unpooled.wrappedBuffer(samplePayload)) + val input = WireFrame(samplePayload) it("should yield decoded input frame") { - WireDecoder().decode(input).test() + createInstance().decode(input).test() .expectNextMatches { it.payloadSize == samplePayload.size } .verifyComplete() } @@ -104,11 +110,11 @@ internal object WireDecoderTest : Spek({ given("valid input with part of next frame") { val input = Unpooled.buffer() - .writeBytes(WireFrame(Unpooled.wrappedBuffer(samplePayload)).encode(alloc)) - .writeBytes(WireFrame(Unpooled.wrappedBuffer(samplePayload)).encode(alloc).slice(0, 3)) + .writeBytes(encoder.encode(WireFrame(samplePayload))) + .writeBytes(encoder.encode(WireFrame(samplePayload)).slice(0, 3)) it("should yield decoded input frame") { - WireDecoder().decode(input).test() + createInstance().decode(input).test() .expectNextMatches { it.payloadSize == samplePayload.size } .verifyComplete() } @@ -120,11 +126,11 @@ internal object WireDecoderTest : Spek({ given("valid input with garbage after it") { val input = Unpooled.buffer() - .writeBytes(WireFrame(Unpooled.wrappedBuffer(samplePayload)).encode(alloc)) + .writeBytes(encoder.encode(WireFrame(samplePayload))) .writeBytes(Unpooled.wrappedBuffer(samplePayload)) it("should yield decoded input frame and error") { - WireDecoder().decode(input).test() + createInstance().decode(input).test() .expectNextMatches { it.payloadSize == samplePayload.size } .verifyError(InvalidWireFrameMarkerException::class.java) } @@ -135,11 +141,11 @@ internal object WireDecoderTest : Spek({ } given("two inputs containing two separate messages") { - val input1 = WireFrame(Unpooled.wrappedBuffer(samplePayload)).encode(alloc) - val input2 = WireFrame(Unpooled.wrappedBuffer(anotherPayload)).encode(alloc) + val input1 = encoder.encode(WireFrame(samplePayload)) + val input2 = encoder.encode(WireFrame(anotherPayload)) it("should yield decoded input frames") { - val cut = WireDecoder() + val cut = createInstance() cut.decode(input1).test() .expectNextMatches { it.payloadSize == samplePayload.size } .verifyComplete() @@ -154,16 +160,12 @@ internal object WireDecoderTest : Spek({ } given("1st input containing 1st frame and 2nd input containing garbage") { - val input1 = WireFrame(Unpooled.wrappedBuffer(samplePayload)).encode(alloc) + val input1 = encoder.encode(WireFrame(samplePayload)) val input2 = Unpooled.wrappedBuffer(anotherPayload) it("should yield decoded input frames") { - val cut = WireDecoder() + val cut = createInstance() cut.decode(input1) - .doOnNext { - // releasing retained payload - it.payload.release() - } .test() .expectNextMatches { it.payloadSize == samplePayload.size } .verifyComplete() @@ -182,8 +184,8 @@ internal object WireDecoderTest : Spek({ given("1st input containing 1st frame + part of 2nd frame and 2nd input containing rest of 2nd frame") { - val frame1 = WireFrame(Unpooled.wrappedBuffer(samplePayload)).encode(alloc) - val frame2 = WireFrame(Unpooled.wrappedBuffer(anotherPayload)).encode(alloc) + val frame1 = encoder.encode(WireFrame(samplePayload)) + val frame2 = encoder.encode(WireFrame(anotherPayload)) val input1 = Unpooled.buffer() .writeBytes(frame1) @@ -191,7 +193,7 @@ internal object WireDecoderTest : Spek({ val input2 = Unpooled.buffer().writeBytes(frame2) it("should yield decoded input frames") { - val cut = WireDecoder() + val cut = createInstance() cut.decode(input1).test() .expectNextMatches { it.payloadSize == samplePayload.size } .verifyComplete() @@ -206,8 +208,8 @@ internal object WireDecoderTest : Spek({ } given("1st input containing part of 1st frame and 2nd input containing rest of 1st + 2nd frame") { - val frame1 = WireFrame(Unpooled.wrappedBuffer(samplePayload)).encode(alloc) - val frame2 = WireFrame(Unpooled.wrappedBuffer(anotherPayload)).encode(alloc) + val frame1 = encoder.encode(WireFrame(samplePayload)) + val frame2 = encoder.encode(WireFrame(anotherPayload)) val input1 = Unpooled.buffer() .writeBytes(frame1, 5) @@ -216,7 +218,7 @@ internal object WireDecoderTest : Spek({ .writeBytes(frame2) it("should yield decoded input frames") { - val cut = WireDecoder() + val cut = createInstance() cut.decode(input1).test() .verifyComplete() cut.decode(input2).test() |