depends_on:
- kafka
volumes:
- - /etc/ves-hv/:/etc/ves-hv/
+ - ./ssl/:/etc/ves-hv/
xnf-simulator:
build:
context: hv-collector-client-simulator
depends_on:
- hv-collector
volumes:
- - /etc/ves-hv/:/etc/ves-hv/
\ No newline at end of file
+ - ./ssl/:/etc/ves-hv/
\ No newline at end of file
LABEL maintainer="Nokia Wroclaw ONAP Team"
WORKDIR /opt/ves-hv-client-simulator
-ENTRYPOINT ["java", "-cp", "*:", "org.onap.dcae.collectors.veshv.main.MainKt"]
+ENTRYPOINT ["java", "-cp", "*:", "org.onap.dcae.collectors.veshv.simulators.xnf.MainKt"]
CMD ["--ves-host", "hv-collector", "--ves-port", "6061"]
COPY target/libs/external/* ./
COPY target/libs/internal/* ./
-COPY target/hv-collector-client-simulator-*.jar ./
\ No newline at end of file
+COPY target/hv-collector-client-simulator-*.jar ./
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-stdlib-jdk8</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-tcnative-boringssl-static</artifactId>
+ <scope>runtime</scope>
+ <classifier>${os.detected.classifier}</classifier>
+ </dependency>
<dependency>
<groupId>com.nhaarman</groupId>
<artifactId>mockito-kotlin</artifactId>
import java.nio.file.Paths
internal object DefaultValues {
- const val MESSAGES_AMOUNT = 1
+ const val MESSAGES_AMOUNT = -1L
const val PRIVATE_KEY_FILE = "/etc/ves-hv/client.key"
const val CERT_FILE = "/etc/ves-hv/client.crt"
const val TRUST_CERT_FILE = "/etc/ves-hv/trust.crt"
val cmdLine = parser.parse(options, args)
val host = cmdLine.stringValue(OPT_VES_HOST)
val port = cmdLine.intValue(OPT_VES_PORT)
- val msgsAmount = cmdLine.intValueOrDefault(OPT_MESSAGES_AMOUNT, DefaultValues.MESSAGES_AMOUNT)
+ val msgsAmount = cmdLine.longValueOrDefault(OPT_MESSAGES_AMOUNT, DefaultValues.MESSAGES_AMOUNT)
return ClientConfiguration(
host,
port,
private fun stringPathToPath(path: String) = Paths.get(File(path).toURI())
- private fun CommandLine.intValueOrDefault(option: Option, default: Int) =
- getOptionValue(option.opt)?.toInt() ?: default
+
+ private fun CommandLine.longValueOrDefault(option: Option, default: Long) =
+ getOptionValue(option.opt)?.toLong() ?: default
private fun CommandLine.intValue(option: Option) =
getOptionValue(option.opt).toInt()
val vesHost: String,
val vesPort: Int,
val security: ClientSecurityConfiguration,
- val messagesAmount: Int)
+ val messagesAmount: Long)
const val DEFAULT_LAST_EPOCH: Long = 120034455
}
- fun createMessageFlux(amount: Int = 1): Flux<WireFrame> =
- Mono.just(createMessage()).repeat(amount.toLong())
+ fun createMessageFlux(amount: Long = -1): Flux<WireFrame> =
+ Mono.fromCallable(this::createMessage).let {
+ if (amount < 0)
+ it.repeat()
+ else
+ it.repeat(amount)
+ }
private fun createMessage(): WireFrame {
.build()
val payload = vesMessageBytes(commonHeader)
- return WireFrame(
- payload = payload,
- mark = 0xFF,
- majorVersion = 1,
- minorVersion = 2,
- payloadSize = payload.readableBytes())
-
-
+ return WireFrame(payload)
}
private fun vesMessageBytes(commonHeader: VesEventV5.VesEvent.CommonEventHeader): ByteBuf {
*/
package org.onap.dcae.collectors.veshv.simulators.xnf.impl
+import io.netty.buffer.Unpooled
import io.netty.handler.ssl.ClientAuth
import io.netty.handler.ssl.SslContext
import io.netty.handler.ssl.SslContextBuilder
import org.onap.dcae.collectors.veshv.utils.logging.Logger
import org.reactivestreams.Publisher
import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
import reactor.ipc.netty.NettyInbound
import reactor.ipc.netty.NettyOutbound
import reactor.ipc.netty.tcp.TcpClient
client.startAndAwait(BiFunction { i, o -> handler(i, o, messages) })
}
- // sending flux with multiple WireFrames not supported yet
private fun handler(nettyInbound: NettyInbound,
nettyOutbound: NettyOutbound,
messages: Flux<WireFrame>): Publisher<Void> {
.subscribe { str -> logger.info("Server response: $str") }
val frames = messages
- .doOnNext { logger.info { "About to send message with ${it.payloadSize} B of payload" } }
.map { it.encode(nettyOutbound.alloc()) }
+ .concatWith(Mono.just(Unpooled.EMPTY_BUFFER))
return nettyOutbound
.options { it.flushOnEach() }
val clientConfig = ArgBasedClientConfiguration.parse(args)
val messageFactory = MessageFactory()
val client = VesHvClient(clientConfig)
- client.send(messageFactory.createMessageFlux(clientConfig.messagesAmount))
+ client.send(messageFactory.createMessageFlux(clientConfig.messagesAmount))
} catch (e: ArgBasedClientConfiguration.WrongArgumentException) {
e.printHelp("java org.onap.dcae.collectors.veshv.main.MainKt")
+ System.exit(1)
} catch (e: Exception) {
logger.error(e.localizedMessage)
logger.debug("An error occurred when starting ves client", e)
+ System.exit(2)
}
}
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.simulators.xnf.impl.MessageFactory
-import kotlin.test.assertEquals
+import reactor.test.test
/**
* @author Jakub Dudycz <jakub.dudycz@nokia.com>
val factory = MessageFactory()
given("no parameters") {
- it("should return flux with one message") {
- val result = factory.createMessageFlux()
-
- assertEquals(1, result.count().block())
+ it("should return infinite flux") {
+ val limit = 1000L
+ factory.createMessageFlux().take(limit).test()
+ .expectNextCount(limit)
+ .verifyComplete()
}
}
given("messages amount") {
it("should return message flux of specified size") {
- val result = factory.createMessageFlux(5)
- assertEquals(5, result.count().block())
+ factory.createMessageFlux(5).test()
+ .expectNextCount(5)
+ .verifyComplete()
}
}
}
package org.onap.dcae.collectors.veshv.boundary
import io.netty.buffer.ByteBuf
+import io.netty.buffer.ByteBufAllocator
import org.onap.dcae.collectors.veshv.model.ServerConfiguration
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
interface Collector {
- fun handleConnection(dataStream: Flux<ByteBuf>): Mono<Void>
+ fun handleConnection(alloc: ByteBufAllocator, dataStream: Flux<ByteBuf>): Mono<Void>
}
typealias CollectorProvider = () -> Collector
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.WireDecoder
+import org.onap.dcae.collectors.veshv.impl.wire.WireDecoder
import reactor.core.publisher.Flux
import java.util.concurrent.atomic.AtomicReference
private fun createVesHvCollector(config: CollectorConfiguration): Collector {
return VesHvCollector(
- WireDecoder(),
+ { WireDecoder(it) },
VesDecoder(),
MessageValidator(),
Router(config.routing),
fun isValid(message: VesMessage): Boolean {
val header = message.header
- return allMandatoryFieldsArePresent(header) && header.domain == CommonEventHeader.Domain.HVRANMEAS
+ val ret = allMandatoryFieldsArePresent(header) && header.domain == CommonEventHeader.Domain.HVRANMEAS
+ return ret
}
private fun allMandatoryFieldsArePresent(header: CommonEventHeader) =
*/
package org.onap.dcae.collectors.veshv.impl
-import com.google.protobuf.InvalidProtocolBufferException
import io.netty.buffer.ByteBuf
import org.onap.dcae.collectors.veshv.model.VesMessage
import org.onap.dcae.collectors.veshv.utils.logging.Logger
*/
internal class VesDecoder {
- fun decode(bb: ByteBuf): VesMessage? =
- try {
- val decodedHeader = VesEvent.parseFrom(bb.nioBuffer()).commonEventHeader
- VesMessage(decodedHeader, bb)
- } catch (ex: InvalidProtocolBufferException) {
- logger.warn { "Dropping incoming message. Invalid protocol buffer: ${ex.message}" }
- logger.debug("Cause", ex)
- null
- }
-
-
- companion object {
- private val logger = Logger(VesDecoder::class)
+ fun decode(bb: ByteBuf): VesMessage {
+ val decodedHeader = VesEvent.parseFrom(bb.nioBuffer()).commonEventHeader
+ return VesMessage(decodedHeader, bb)
}
}
package org.onap.dcae.collectors.veshv.impl
import io.netty.buffer.ByteBuf
+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.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 wireDecoder: WireDecoder,
+ private val wireDecoderSupplier: (ByteBufAllocator) -> WireDecoder,
private val protobufDecoder: VesDecoder,
private val validator: MessageValidator,
private val router: Router,
private val sink: Sink) : Collector {
- override fun handleConnection(dataStream: Flux<ByteBuf>): Mono<Void> =
- dataStream
- .doOnNext(this::logIncomingMessage)
- .flatMap(this::decodeWire)
- .doOnNext(this::logDecodedWireMessage)
- .flatMap(this::decodeProtobuf)
- .filter(this::validate)
- .flatMap(this::findRoute)
- .compose(sink::send)
- .doOnNext(this::releaseMemory)
- .then()
- private fun logIncomingMessage(wire: ByteBuf) {
- logger.debug { "Got message with total ${wire.readableBytes()} B"}
- }
-
- private fun logDecodedWireMessage(payload: ByteBuf) {
- logger.debug { "Wire payload size: ${payload.readableBytes()} B"}
- }
-
- private fun decodeWire(wire: ByteBuf) = omitWhenNull(wire, wireDecoder::decode)
-
- private fun decodeProtobuf(protobuf: ByteBuf) = releaseWhenNull(protobuf, protobufDecoder::decode)
+ override fun handleConnection(alloc: ByteBufAllocator, dataStream: Flux<ByteBuf>): Mono<Void> =
+ wireDecoderSupplier(alloc).let { wireDecoder ->
+ dataStream
+ .concatMap(wireDecoder::decode)
+ .filter(WireFrame::isValid)
+ .map(WireFrame::payload)
+ .map(protobufDecoder::decode)
+ .filter(this::validate)
+ .flatMap(this::findRoute)
+ .compose(sink::send)
+ .doOnNext(this::releaseMemory)
+ .doOnTerminate { releaseBuffersMemory(wireDecoder) }
+ .then()
+ }
private fun validate(msg: VesMessage): Boolean {
val valid = validator.isValid(msg)
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 <T, V>omitWhenNull(input: T, mapper: (T) -> V?): Mono<V> = Mono.justOrEmpty(mapper(input))
-
- private fun <T>releaseWhenNull(input: ByteBuf, mapper: (ByteBuf) -> T?): Mono<T> {
- val result = mapper(input)
- return if (result == null) {
- input.release()
- Mono.empty()
- } else {
- Mono.just(result)
- }
+ private fun releaseBuffersMemory(wireDecoder: WireDecoder) {
+ wireDecoder.release()
}
+ private fun <T, V> omitWhenNull(input: T, mapper: (T) -> V?): Mono<V> = Mono.justOrEmpty(mapper(input))
+
companion object {
val logger = Logger(VesHvCollector::class)
}
*/
package org.onap.dcae.collectors.veshv.impl.adapters
-import org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG
-import org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG
-import org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG
-import org.apache.kafka.common.serialization.ByteBufferSerializer
-import org.apache.kafka.common.serialization.StringSerializer
import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
-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.dcae.collectors.veshv.model.RoutedMessage
-import org.onap.dcae.collectors.veshv.model.VesMessage
-import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
import reactor.core.publisher.Flux
import reactor.ipc.netty.http.client.HttpClient
-import reactor.kafka.sender.KafkaSender
-import reactor.kafka.sender.SenderOptions
-import java.nio.ByteBuffer
/**
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
override fun invoke() = Flux.just(config)
}
- private class KafkaSinkProvider : SinkProvider {
- override fun invoke(config: CollectorConfiguration): Sink {
- val sender = KafkaSender.create(
- SenderOptions.create<CommonEventHeader, ByteBuffer>()
- .producerProperty(BOOTSTRAP_SERVERS_CONFIG, config.kafkaBootstrapServers)
- .producerProperty(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java)
- .producerProperty(VALUE_SERIALIZER_CLASS_CONFIG, ByteBufferSerializer::class.java))
- return KafkaSink(sender)
- }
- }
-
-
- private class LoggingSinkProvider : SinkProvider {
- override fun invoke(config: CollectorConfiguration): Sink {
- return object : Sink {
- private val logger = Logger(LoggingSinkProvider::class)
- override fun send(messages: Flux<RoutedMessage>): Flux<VesMessage> =
- messages
- .doOnNext { msg ->
- logger.info { "Message routed to ${msg.topic}" }
- }
- .map { it.message }
-
- }
- }
- }
-
fun consulConfigurationProvider(url: String): ConfigurationProvider =
ConsulConfigurationProvider(url, httpAdapter())
--- /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.impl.adapters
+
+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 reactor.kafka.sender.KafkaSender
+import reactor.kafka.sender.SenderOptions
+import java.nio.ByteBuffer
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+internal class KafkaSinkProvider : SinkProvider {
+ override fun invoke(config: CollectorConfiguration): Sink {
+ return KafkaSink(KafkaSender.create(constructSenderOptions(config)))
+ }
+
+ private fun constructSenderOptions(config: CollectorConfiguration) =
+ SenderOptions.create<VesEventV5.VesEvent.CommonEventHeader, ByteBuffer>()
+ .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)
+
+}
--- /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.impl.adapters
+
+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.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 java.util.concurrent.atomic.AtomicLong
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+internal class LoggingSinkProvider : SinkProvider {
+
+ override fun invoke(config: CollectorConfiguration): Sink {
+ return object : Sink {
+ private val logger = Logger(LoggingSinkProvider::class)
+ private val totalMessages = AtomicLong()
+ private val totalBytes = AtomicLong()
+
+ override fun send(messages: Flux<RoutedMessage>): Flux<VesMessage> =
+ messages
+ .doOnNext(this::logMessage)
+ .map { it.message }
+
+ private fun logMessage(msg: RoutedMessage) {
+ val msgs = totalMessages.addAndGet(1)
+ val bytes = totalBytes.addAndGet(msg.message.rawMessage.readableBytes().toLong())
+ val logMessageSupplier = { "Message routed to ${msg.topic}. Total = $msgs ($bytes B)" }
+ if (msgs % INFO_LOGGING_FREQ == 0L)
+ logger.info(logMessageSupplier)
+ else
+ logger.trace(logMessageSupplier)
+ }
+
+ }
+ }
+
+ companion object {
+ const val INFO_LOGGING_FREQ = 100_000
+ }
+}
*/
package org.onap.dcae.collectors.veshv.impl.socket
+import io.netty.buffer.ByteBuf
import org.onap.dcae.collectors.veshv.boundary.CollectorProvider
import org.onap.dcae.collectors.veshv.boundary.Server
import org.onap.dcae.collectors.veshv.model.ServerConfiguration
private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> {
logger.debug("Got connection")
+ nettyOutbound.alloc()
val sendHello = nettyOutbound
.options { it.flushOnEach() }
.sendString(Mono.just("ONAP_VES_HV/0.1\n"))
.then()
- val handleIncomingMessages = collectorProvider().handleConnection(nettyInbound.receive())
+ val handleIncomingMessages = collectorProvider()
+ .handleConnection(nettyInbound.context().channel().alloc(), nettyInbound.receive().retain())
return sendHello.then(handleIncomingMessages)
}
-
companion object {
private val logger = Logger(NettyTcpServer::class)
}
--- /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.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.utils.logging.Logger
+import reactor.core.publisher.Flux
+import reactor.core.publisher.FluxSink
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.function.Consumer
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+internal class StreamBufferEmitter(
+ private val streamBuffer: CompositeByteBuf,
+ private val newFrame: ByteBuf)
+ : Consumer<FluxSink<WireFrame>> {
+
+ private val subscribed = AtomicBoolean(false)
+
+ override fun accept(sink: FluxSink<WireFrame>) {
+ when {
+
+ subscribed.getAndSet(true) ->
+ sink.error(IllegalStateException("Wire frame emitter supports only one subscriber"))
+
+ newFrame.readableBytes() == 0 -> {
+ logger.trace { "Discarding empty buffer" }
+ newFrame.release()
+ sink.complete()
+ }
+
+ else -> {
+ streamBuffer.addComponent(INCREASE_WRITER_INDEX, newFrame)
+ sink.onDispose {
+ logger.debug("Disposing read components")
+ streamBuffer.discardReadComponents()
+ }
+ sink.onRequest { requestedFrameCount ->
+ WireFrameSink(streamBuffer, sink, requestedFrameCount).handleSubscriber()
+ }
+ }
+ }
+ }
+
+ companion object {
+ fun createFlux(streamBuffer: CompositeByteBuf, newFrame: ByteBuf): Flux<WireFrame> =
+ Flux.create(StreamBufferEmitter(streamBuffer, newFrame))
+
+ private const val INCREASE_WRITER_INDEX = true
+ private val logger = Logger(StreamBufferEmitter::class)
+ }
+}
--- /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.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.impl.VesHvCollector
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import reactor.core.publisher.Flux
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+internal class WireDecoder(alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT) {
+ private val streamBuffer = alloc.compositeBuffer()
+
+ fun decode(byteBuf: ByteBuf): Flux<WireFrame> = StreamBufferEmitter.createFlux(streamBuffer, byteBuf)
+ .doOnSubscribe { logIncomingMessage(byteBuf) }
+ .doOnNext(this::logDecodedWireMessage)
+
+ fun release() {
+ streamBuffer.release()
+ }
+
+
+ private fun logIncomingMessage(wire: ByteBuf) {
+ logger.trace { "Got message with total size of ${wire.readableBytes()} B" }
+ }
+
+ private fun logDecodedWireMessage(wire: WireFrame) {
+ logger.trace { "Wire payload size: ${wire.payloadSize} B." }
+ }
+
+ companion object {
+ val logger = Logger(VesHvCollector::class)
+ }
+}
--- /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.impl.wire
+
+import io.netty.buffer.ByteBuf
+import org.onap.dcae.collectors.veshv.domain.WireFrame
+import org.onap.dcae.collectors.veshv.domain.exceptions.MissingWireFrameBytesException
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import reactor.core.publisher.FluxSink
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+internal class WireFrameSink(
+ private val streamBuffer: ByteBuf,
+ private val sink: FluxSink<WireFrame>,
+ private val requestedFrameCount: Long) {
+
+ fun handleSubscriber() {
+ 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()
+ }
+ } catch (ex: Exception) {
+ sink.error(ex)
+ }
+
+ logger.trace { "Decoder buffer capacity after decoding frame: ${streamBuffer.capacity()}" }
+
+ }
+
+ private fun pushAvailableFrames() {
+ var nextFrame = decodeFirstFrameFromBuffer()
+ while (nextFrame != null && !sink.isCancelled) {
+ sink.next(nextFrame)
+ nextFrame = decodeFirstFrameFromBuffer()
+ }
+ sink.complete()
+ }
+
+ private fun pushUpToNumberOfFrames() {
+ var nextFrame = decodeFirstFrameFromBuffer()
+ var remaining = requestedFrameCount
+ loop@ while (nextFrame != null && !sink.isCancelled) {
+ sink.next(nextFrame)
+ if (--remaining > 0) {
+ nextFrame = decodeFirstFrameFromBuffer()
+ } else {
+ break@loop
+ }
+ }
+ if (remaining > 0 && nextFrame == null) {
+ sink.complete()
+ }
+ }
+
+ private fun decodeFirstFrameFromBuffer(): WireFrame? =
+ try {
+ WireFrame.decodeFirst(streamBuffer)
+ } catch (ex: MissingWireFrameBytesException) {
+ logger.debug { "${ex.message} - waiting for more data" }
+ null
+ }
+
+ companion object {
+ private val logger = Logger(WireFrameSink::class)
+ }
+}
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
on("invalid ves hv message bytes") {
val rawMessageBytes = wrappedBuffer("ala ma kota".toByteArray(Charset.defaultCharset()))
- it("should return empty result") {
- assertThat(cut.decode(rawMessageBytes)).isNull()
+ it("should throw error") {
+ assertThatExceptionOfType(InvalidProtocolBufferException::class.java)
+ .isThrownBy { cut.decode(rawMessageBytes) }
}
}
}
+++ /dev/null
-package org.onap.dcae.collectors.veshv.impl
-
-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.given
-import org.jetbrains.spek.api.dsl.it
-import org.onap.dcae.collectors.veshv.domain.WireFrame
-
-/**
- * @author Piotr Jaszczyk <piotr.jaszczyk></piotr.jaszczyk>@nokia.com>
- * @since May 2018
- */
-internal object WireDecoderTest : Spek({
- describe("decoding wire protocol") {
- val cut = WireDecoder()
-
- fun decode(frame: WireFrame) =
- cut.decode(
- frame.encode(UnpooledByteBufAllocator.DEFAULT))
-
- given("empty input") {
- val input = Unpooled.EMPTY_BUFFER
-
- it("should yield empty result") {
- assertThat(cut.decode(input)).isNull()
- }
- }
-
- given("input without 0xFF first byte") {
- val input = WireFrame(
- payload = Unpooled.EMPTY_BUFFER,
- mark = 0x10,
- majorVersion = 1,
- minorVersion = 2,
- payloadSize = 0)
-
- it("should yield empty result") {
- assertThat(decode(input)).isNull()
- }
- }
-
- given("input with unsupported major version") {
- val input = WireFrame(
- payload = Unpooled.EMPTY_BUFFER,
- mark = 0xFF,
- majorVersion = 100,
- minorVersion = 2,
- payloadSize = 0)
-
- it("should yield empty result") {
- assertThat(decode(input)).isNull()
- }
- }
-
- given("input with too small payload size") {
- val input = WireFrame(
- payload = Unpooled.wrappedBuffer(byteArrayOf(1, 2, 3)),
- mark = 0xFF,
- majorVersion = 1,
- minorVersion = 0,
- payloadSize = 1)
-
- it("should yield empty result") {
- assertThat(decode(input)).isNull()
- }
- }
-
- given("input with too big payload size") {
- val input = WireFrame(
- payload = Unpooled.wrappedBuffer(byteArrayOf(1, 2, 3)),
- mark = 0xFF,
- majorVersion = 1,
- minorVersion = 0,
- payloadSize = 8)
-
- it("should yield empty result") {
- assertThat(decode(input)).isNull()
- }
- }
-
- given("valid input") {
- val payload = byteArrayOf(6, 9, 8, 6)
- val input = WireFrame(
- payload = Unpooled.wrappedBuffer(payload),
- mark = 0xFF,
- majorVersion = 1,
- minorVersion = 0,
- payloadSize = payload.size)
-
-
- it("should yield Google Protocol Buffers payload") {
- val result = decode(input)!!
-
- val actualPayload = ByteArray(result.readableBytes())
- result.readBytes(actualPayload)
-
- assertThat(actualPayload).containsExactly(*payload)
- }
- }
- }
-})
--- /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.impl.wire
+
+import io.netty.buffer.ByteBuf
+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.given
+import org.jetbrains.spek.api.dsl.it
+import org.onap.dcae.collectors.veshv.domain.WireFrame
+import org.onap.dcae.collectors.veshv.domain.exceptions.InvalidWireFrameMarkerException
+import reactor.test.test
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk></piotr.jaszczyk>@nokia.com>
+ * @since May 2018
+ */
+internal object WireDecoderTest : 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))
+
+ fun verifyMemoryReleased(vararg byteBuffers: ByteBuf) {
+ for (bb in byteBuffers) {
+ assertThat(bb.refCnt())
+ .describedAs("should be released: $bb ref count")
+ .isEqualTo(0)
+ }
+ }
+
+ fun verifyMemoryNotReleased(vararg byteBuffers: ByteBuf) {
+ for (bb in byteBuffers) {
+ assertThat(bb.refCnt())
+ .describedAs("should not be released: $bb ref count")
+ .isEqualTo(1)
+ }
+ }
+
+ describe("decoding wire protocol") {
+ given("empty input") {
+ val input = Unpooled.EMPTY_BUFFER
+
+ it("should yield empty result") {
+ WireDecoder().decode(input).test().verifyComplete()
+ }
+ }
+
+ given("input with no readable bytes") {
+ val input = Unpooled.wrappedBuffer(byteArrayOf(0x00)).readerIndex(1)
+
+ it("should yield empty result") {
+ WireDecoder().decode(input).test().verifyComplete()
+ }
+
+ it("should release memory") {
+ verifyMemoryReleased(input)
+ }
+ }
+
+ given("invalid input (not starting with marker)") {
+ val input = Unpooled.wrappedBuffer(samplePayload)
+
+ it("should yield error") {
+ WireDecoder().decode(input).test()
+ .verifyError(InvalidWireFrameMarkerException::class.java)
+ }
+
+ it("should leave memory unreleased") {
+ verifyMemoryNotReleased(input)
+ }
+ }
+
+ given("valid input") {
+ val input = WireFrame(Unpooled.wrappedBuffer(samplePayload))
+
+ it("should yield decoded input frame") {
+ WireDecoder().decode(input).test()
+ .expectNextMatches { it.payloadSize == samplePayload.size }
+ .verifyComplete()
+ }
+ }
+
+ 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))
+
+ it("should yield decoded input frame") {
+ WireDecoder().decode(input).test()
+ .expectNextMatches { it.payloadSize == samplePayload.size }
+ .verifyComplete()
+ }
+
+ it("should leave memory unreleased") {
+ verifyMemoryNotReleased(input)
+ }
+ }
+
+ given("valid input with garbage after it") {
+ val input = Unpooled.buffer()
+ .writeBytes(WireFrame(Unpooled.wrappedBuffer(samplePayload)).encode(alloc))
+ .writeBytes(Unpooled.wrappedBuffer(samplePayload))
+
+ it("should yield decoded input frame and error") {
+ WireDecoder().decode(input).test()
+ .expectNextMatches { it.payloadSize == samplePayload.size }
+ .verifyError(InvalidWireFrameMarkerException::class.java)
+ }
+
+ it("should leave memory unreleased") {
+ verifyMemoryNotReleased(input)
+ }
+ }
+
+ given("two inputs containing two separate messages") {
+ val input1 = WireFrame(Unpooled.wrappedBuffer(samplePayload)).encode(alloc)
+ val input2 = WireFrame(Unpooled.wrappedBuffer(anotherPayload)).encode(alloc)
+
+ it("should yield decoded input frames") {
+ val cut = WireDecoder()
+ cut.decode(input1).test()
+ .expectNextMatches { it.payloadSize == samplePayload.size }
+ .verifyComplete()
+ cut.decode(input2).test()
+ .expectNextMatches { it.payloadSize == anotherPayload.size }
+ .verifyComplete()
+ }
+
+ it("should release memory") {
+ verifyMemoryReleased(input1, input2)
+ }
+ }
+
+ given("1st input containing 1st frame and 2nd input containing garbage") {
+ val input1 = WireFrame(Unpooled.wrappedBuffer(samplePayload)).encode(alloc)
+ val input2 = Unpooled.wrappedBuffer(anotherPayload)
+
+ it("should yield decoded input frames") {
+ val cut = WireDecoder()
+ cut.decode(input1)
+ .doOnNext {
+ // releasing retained payload
+ it.payload.release()
+ }
+ .test()
+ .expectNextMatches { it.payloadSize == samplePayload.size }
+ .verifyComplete()
+ cut.decode(input2).test()
+ .verifyError(InvalidWireFrameMarkerException::class.java)
+ }
+
+ it("should release memory for 1st input") {
+ verifyMemoryReleased(input1)
+ }
+
+ it("should leave memory unreleased for 2nd input") {
+ verifyMemoryNotReleased(input2)
+ }
+ }
+
+
+ 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 input1 = Unpooled.buffer()
+ .writeBytes(frame1)
+ .writeBytes(frame2, 3)
+ val input2 = Unpooled.buffer().writeBytes(frame2)
+
+ it("should yield decoded input frames") {
+ val cut = WireDecoder()
+ cut.decode(input1).test()
+ .expectNextMatches { it.payloadSize == samplePayload.size }
+ .verifyComplete()
+ cut.decode(input2).test()
+ .expectNextMatches { it.payloadSize == anotherPayload.size }
+ .verifyComplete()
+ }
+
+ it("should release memory") {
+ verifyMemoryReleased(input1, input2)
+ }
+ }
+
+ 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 input1 = Unpooled.buffer()
+ .writeBytes(frame1, 5)
+ val input2 = Unpooled.buffer()
+ .writeBytes(frame1)
+ .writeBytes(frame2)
+
+ it("should yield decoded input frames") {
+ val cut = WireDecoder()
+ cut.decode(input1).test()
+ .verifyComplete()
+ cut.decode(input2).test()
+ .expectNextMatches { it.payloadSize == samplePayload.size }
+ .expectNextMatches { it.payloadSize == anotherPayload.size }
+ .verifyComplete()
+ }
+
+ it("should release memory") {
+ verifyMemoryReleased(input1, input2)
+ }
+ }
+ }
+})
</rollingPolicy>
</appender>
- <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
+ <logger name="org.onap.dcae.collectors.veshv" level="TRACE"/>
<root level="INFO">
<appender-ref ref="CONSOLE"/>
package org.onap.dcae.collectors.veshv.tests.component
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.SinkProvider
import org.onap.dcae.collectors.veshv.model.RoutedMessage
import org.onap.dcae.collectors.veshv.factory.CollectorFactory
import org.onap.dcae.collectors.veshv.tests.fakes.FakeConfigurationProvider
import org.onap.dcae.collectors.veshv.tests.fakes.FakeSink
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import reactor.core.Exceptions
import reactor.core.publisher.Flux
import java.time.Duration
internal class Sut {
val configurationProvider = FakeConfigurationProvider()
val sink = FakeSink()
+ val alloc = UnpooledByteBufAllocator.DEFAULT
private val collectorFactory = CollectorFactory(configurationProvider, SinkProvider.just(sink))
val collectorProvider = collectorFactory.createVesHvCollectorProvider()
get() = collectorProvider()
fun handleConnection(vararg packets: ByteBuf): List<RoutedMessage> {
- collector.handleConnection(Flux.fromArray(packets)).block(Duration.ofSeconds(10))
-
+ 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)
+ }
}
*/
package org.onap.dcae.collectors.veshv.tests.component
+import com.google.protobuf.InvalidProtocolBufferException
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.domain.exceptions.InvalidWireFrameMarkerException
+import org.onap.dcae.collectors.veshv.domain.exceptions.WireFrameDecodingException
import org.onap.dcae.collectors.veshv.tests.fakes.HVRANMEAS_TOPIC
import org.onap.dcae.collectors.veshv.tests.fakes.basicConfiguration
import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
.describedAs("should send all events")
.hasSize(2)
}
+ }
+
+ describe("Memory management") {
- system("should release memory for each incoming message") { sut ->
+ system("should release memory for each handled and dropped message") { sut ->
sut.configurationProvider.updateConfiguration(basicConfiguration)
+ val validMessage = vesMessage(Domain.HVRANMEAS)
val msgWithInvalidDomain = vesMessage(Domain.OTHER)
- val msgWithInvalidPayload = invalidVesMessage()
val msgWithInvalidFrame = invalidWireFrame()
- val validMessage = vesMessage(Domain.HVRANMEAS)
- val refCntBeforeSending = msgWithInvalidDomain.refCnt()
+ val expectedRefCnt = 0
+
+ val (handledEvents, exception) = sut.handleConnectionReturningError(
+ validMessage, msgWithInvalidDomain, msgWithInvalidFrame)
- sut.handleConnection(msgWithInvalidDomain, msgWithInvalidPayload, msgWithInvalidFrame, validMessage)
+ assertThat(handledEvents).hasSize(1)
+ assertThat(exception).isNull()
+ assertThat(validMessage.refCnt())
+ .describedAs("handled message should be released")
+ .isEqualTo(expectedRefCnt)
assertThat(msgWithInvalidDomain.refCnt())
.describedAs("message with invalid domain should be released")
- .isEqualTo(refCntBeforeSending)
- assertThat(msgWithInvalidPayload.refCnt())
- .describedAs("message with invalid payload should be released")
- .isEqualTo(refCntBeforeSending)
+ .isEqualTo(expectedRefCnt)
assertThat(msgWithInvalidFrame.refCnt())
.describedAs("message with invalid frame should be released")
- .isEqualTo(refCntBeforeSending)
+ .isEqualTo(expectedRefCnt)
+
+ }
+
+ system("should release memory for each message with invalid payload") { sut ->
+ sut.configurationProvider.updateConfiguration(basicConfiguration)
+ val validMessage = vesMessage(Domain.HVRANMEAS)
+ val msgWithInvalidPayload = invalidVesMessage()
+ val expectedRefCnt = 0
+
+ val (handledEvents, exception) = sut.handleConnectionReturningError(
+ validMessage, msgWithInvalidPayload)
+
+ assertThat(handledEvents).hasSize(1)
+ assertThat(exception?.cause).isInstanceOf(InvalidProtocolBufferException::class.java)
+
+ assertThat(validMessage.refCnt())
+ .describedAs("handled message should be released")
+ .isEqualTo(expectedRefCnt)
+ assertThat(msgWithInvalidPayload.refCnt())
+ .describedAs("message with invalid payload should be released")
+ .isEqualTo(expectedRefCnt)
+
+ }
+
+ system("should release memory for each message with garbage frame") { sut ->
+ sut.configurationProvider.updateConfiguration(basicConfiguration)
+ val validMessage = vesMessage(Domain.HVRANMEAS)
+ val msgWithGarbageFrame = garbageFrame()
+ val expectedRefCnt = 0
+
+ val (handledEvents, exception) = sut.handleConnectionReturningError(
+ validMessage, msgWithGarbageFrame)
+
+ assertThat(handledEvents).hasSize(1)
+ assertThat(exception?.cause)
+ .isInstanceOf(InvalidWireFrameMarkerException::class.java)
+
assertThat(validMessage.refCnt())
.describedAs("handled message should be released")
- .isEqualTo(refCntBeforeSending)
+ .isEqualTo(expectedRefCnt)
+ assertThat(msgWithGarbageFrame.refCnt())
+ .describedAs("message with garbage frame should be released")
+ .isEqualTo(expectedRefCnt)
+
}
}
}
+fun garbageFrame() = alocator.buffer().run {
+ writeBytes("the meaning of life is &@)(*_!".toByteArray())
+}
+
fun invalidWireFrame() = alocator.buffer().run {
writeByte(0xFF)
writeByte(1)
.setCommonEventHeader(
CommonEventHeader.getDefaultInstance().toBuilder()
.setVersion("1.0")
+ .setEventName("xyz")
.setEventId(id)
.setDomain(domain)
.setEventName("Sample event name")
.setSequence(1))
.setHvRanMeasFields(ByteString.EMPTY)
.build()
-
-
-
</rollingPolicy>
</appender>
- <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
+ <logger name="org.onap.dcae.collectors.veshv" level="TRACE"/>
<root level="INFO">
<appender-ref ref="CONSOLE"/>
import io.netty.buffer.ByteBuf
import io.netty.buffer.ByteBufAllocator
+import org.onap.dcae.collectors.veshv.domain.exceptions.EmptyWireFrameException
+import org.onap.dcae.collectors.veshv.domain.exceptions.InvalidWireFrameMarkerException
+import org.onap.dcae.collectors.veshv.domain.exceptions.MissingWireFrameBytesException
/**
* Wire frame structure is presented bellow. All fields are in network byte order (big-endian).
* @since May 2018
*/
data class WireFrame(val payload: ByteBuf,
- val mark: Short,
val majorVersion: Short,
val minorVersion: Short,
val payloadSize: Int) {
+ constructor(payload: ByteBuf) : this(payload, 1, 0, payload.readableBytes())
+
fun isValid(): Boolean =
- mark == FF_BYTE
- && majorVersion == SUPPORTED_MAJOR_VERSION
+ majorVersion == SUPPORTED_MAJOR_VERSION
&& payload.readableBytes() == payloadSize
fun encode(allocator: ByteBufAllocator): ByteBuf {
val bb = allocator.buffer(HEADER_SIZE + payload.readableBytes())
- bb.writeByte(mark.toInt())
+ bb.writeByte(MARKER_BYTE.toInt())
bb.writeByte(majorVersion.toInt())
bb.writeByte(minorVersion.toInt())
bb.writeInt(payloadSize)
}
companion object {
- fun decode(byteBuf: ByteBuf): WireFrame {
- val mark = byteBuf.readUnsignedByte()
+ fun decodeFirst(byteBuf: ByteBuf): WireFrame {
+ verifyNotEmpty(byteBuf)
+ byteBuf.markReaderIndex()
+
+ verifyMarker(byteBuf)
+ verifyMinimumSize(byteBuf)
+
val majorVersion = byteBuf.readUnsignedByte()
val minorVersion = byteBuf.readUnsignedByte()
- val payloadSize = byteBuf.readInt()
- val payload = byteBuf.retainedSlice()
+ val payloadSize = verifyPayloadSize(byteBuf)
+
+ val payload = byteBuf.retainedSlice(byteBuf.readerIndex(), payloadSize)
+ byteBuf.readerIndex(byteBuf.readerIndex() + payloadSize)
+
+ return WireFrame(payload, majorVersion, minorVersion, payloadSize)
+ }
+
+ private fun verifyPayloadSize(byteBuf: ByteBuf): Int =
+ byteBuf.readInt().let { payloadSize ->
+ if (byteBuf.readableBytes() < payloadSize) {
+ byteBuf.resetReaderIndex()
+ throw MissingWireFrameBytesException("readable bytes < payload size")
+ } else {
+ payloadSize
+ }
+ }
+
+ private fun verifyMinimumSize(byteBuf: ByteBuf) {
+ if (byteBuf.readableBytes() < HEADER_SIZE) {
+ byteBuf.resetReaderIndex()
+ throw MissingWireFrameBytesException("readable bytes < header size")
+ }
+ }
+
+ private fun verifyMarker(byteBuf: ByteBuf) {
+ val mark = byteBuf.readUnsignedByte()
+ if (mark != MARKER_BYTE) {
+ byteBuf.resetReaderIndex()
+ throw InvalidWireFrameMarkerException(mark)
+ }
+ }
- return WireFrame(payload, mark, majorVersion, minorVersion, payloadSize)
+ private fun verifyNotEmpty(byteBuf: ByteBuf) {
+ if (byteBuf.readableBytes() < 1) {
+ throw EmptyWireFrameException()
+ }
}
- private const val HEADER_SIZE =
+ const val HEADER_SIZE =
3 * java.lang.Byte.BYTES +
- 1 * java.lang.Integer.BYTES
- private const val FF_BYTE: Short = 0xFF
- private const val SUPPORTED_MAJOR_VERSION: Short = 1
+ 1 * java.lang.Integer.BYTES
+ const val MARKER_BYTE: Short = 0xFF
+ const val SUPPORTED_MAJOR_VERSION: Short = 1
}
}
--- /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.domain.exceptions
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+class EmptyWireFrameException : MissingWireFrameBytesException("wire frame was empty (readable bytes == 0)")
* limitations under the License.
* ============LICENSE_END=========================================================
*/
-package org.onap.dcae.collectors.veshv.impl
+package org.onap.dcae.collectors.veshv.domain.exceptions
-import io.netty.buffer.ByteBuf
import org.onap.dcae.collectors.veshv.domain.WireFrame
-import org.onap.dcae.collectors.veshv.utils.logging.Logger
/**
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
- * @since May 2018
+ * @since June 2018
*/
-internal class WireDecoder {
- fun decode(byteBuf: ByteBuf): ByteBuf? =
- try {
- WireFrame.decode(byteBuf)
- .takeIf { it.isValid() }
- .let { it?.payload }
- } catch (ex: IndexOutOfBoundsException) {
- logger.debug { "Wire protocol frame could not be decoded - input is too small" }
- null
- }
-
- companion object {
- private val logger = Logger(WireDecoder::class)
- }
-}
+class InvalidWireFrameMarkerException(actualMarker: Short) : WireFrameDecodingException(
+ "Invalid start of frame. Expected 0x%02X, but was 0x%02X".format(WireFrame.MARKER_BYTE, actualMarker))
--- /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.domain.exceptions
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+open class MissingWireFrameBytesException(msg: String) : WireFrameDecodingException(msg)
--- /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.domain.exceptions
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+open class WireFrameDecodingException(msg: String) : Exception(msg)
+/*
+ * ============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.domain
-import io.netty.buffer.ByteBufAllocator
import io.netty.buffer.Unpooled
+import io.netty.buffer.UnpooledByteBufAllocator
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.describe
+import org.jetbrains.spek.api.dsl.given
import org.jetbrains.spek.api.dsl.it
+import org.onap.dcae.collectors.veshv.domain.exceptions.InvalidWireFrameMarkerException
+import org.onap.dcae.collectors.veshv.domain.exceptions.MissingWireFrameBytesException
+import java.nio.charset.Charset
/**
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
* @since June 2018
*/
object WireFrameTest : Spek({
- describe("Wire Frame codec") {
- describe("encode-decode methods' compatibility") {
- val payloadContent = "test"
- val payload = Unpooled.wrappedBuffer(payloadContent.toByteArray(Charsets.US_ASCII))
- val frame = WireFrame(payload = payload,
- majorVersion = 1,
+ val payloadAsString = "coffeebabe"
+
+ fun createSampleFrame() =
+ WireFrame(Unpooled.wrappedBuffer(payloadAsString.toByteArray(Charset.defaultCharset())))
+
+ fun encodeSampleFrame() =
+ createSampleFrame().let {
+ Unpooled.buffer()
+ .writeBytes(it.encode(UnpooledByteBufAllocator.DEFAULT))
+
+ }
+
+ describe("Wire Frame invariants") {
+
+ given("input with unsupported major version") {
+ val input = WireFrame(
+ payload = Unpooled.EMPTY_BUFFER,
+ majorVersion = 100,
minorVersion = 2,
- mark = 0xFF,
- payloadSize = payload.readableBytes())
+ payloadSize = 0)
+
+ it("should fail validation") {
+ assertThat(input.isValid()).isFalse()
+ }
+ }
+
+ given("input with too small payload size") {
+ val input = WireFrame(
+ payload = Unpooled.wrappedBuffer(byteArrayOf(1, 2, 3)),
+ majorVersion = 1,
+ minorVersion = 0,
+ payloadSize = 1)
+
+ it("should fail validation") {
+ assertThat(input.isValid()).isFalse()
+ }
+ }
+
+ given("input with too big payload size") {
+ val input = WireFrame(
+ payload = Unpooled.wrappedBuffer(byteArrayOf(1, 2, 3)),
+ majorVersion = 1,
+ minorVersion = 0,
+ payloadSize = 8)
+
+ it("should fail validation") {
+ assertThat(input.isValid()).isFalse()
+ }
+ }
+
+ given("valid input") {
+ val payload = byteArrayOf(6, 9, 8, 6)
+ val input = WireFrame(
+ payload = Unpooled.wrappedBuffer(payload),
+ majorVersion = 1,
+ minorVersion = 0,
+ payloadSize = payload.size)
+
+ it("should pass validation") {
+ assertThat(input.isValid()).isTrue()
+ }
+ }
- val encoded = frame.encode(ByteBufAllocator.DEFAULT)
- val decoded = WireFrame.decode(encoded)
+
+ }
+
+ describe("Wire Frame codec") {
+
+ describe("encode-decode methods' compatibility") {
+ val frame = createSampleFrame()
+ val encoded = encodeSampleFrame()
+ val decoded = WireFrame.decodeFirst(encoded)
it("should decode major version") {
assertThat(decoded.majorVersion).isEqualTo(frame.majorVersion)
assertThat(decoded.minorVersion).isEqualTo(frame.minorVersion)
}
- it("should decode mark") {
- assertThat(decoded.mark).isEqualTo(frame.mark)
- }
-
it("should decode payload size") {
assertThat(decoded.payloadSize).isEqualTo(frame.payloadSize)
}
it("should decode payload") {
- assertThat(decoded.payload.toString(Charsets.US_ASCII))
- .isEqualTo(payloadContent)
+ assertThat(decoded.payload.toString(Charset.defaultCharset()))
+ .isEqualTo(payloadAsString)
}
it("should retain decoded payload") {
assertThat(decoded.payload.refCnt()).isEqualTo(1)
}
}
+
+ describe("TCP framing") {
+ // see "Dealing with a Stream-based Transport" on http://netty.io/wiki/user-guide-for-4.x.html#wiki-h3-11
+
+ it("should decode message leaving rest unread") {
+ val buff = Unpooled.buffer()
+ .writeBytes(encodeSampleFrame())
+ .writeByte(0xAA)
+ val decoded = WireFrame.decodeFirst(buff)
+
+ assertThat(decoded.isValid()).describedAs("should be valid").isTrue()
+ assertThat(buff.readableBytes()).isEqualTo(1)
+ }
+
+ it("should throw exception when not even header fits") {
+ val buff = Unpooled.buffer()
+ .writeByte(0xFF)
+
+ assertThatExceptionOfType(MissingWireFrameBytesException::class.java)
+ .isThrownBy { WireFrame.decodeFirst(buff) }
+ }
+
+ it("should throw exception when first byte is not 0xFF but length looks ok") {
+ val buff = Unpooled.buffer()
+ .writeByte(0xAA)
+ .writeBytes("some garbage".toByteArray())
+
+ assertThatExceptionOfType(InvalidWireFrameMarkerException::class.java)
+ .isThrownBy { WireFrame.decodeFirst(buff) }
+ }
+
+ it("should throw exception when first byte is not 0xFF and length is to short") {
+ val buff = Unpooled.buffer()
+ .writeByte(0xAA)
+
+ assertThatExceptionOfType(InvalidWireFrameMarkerException::class.java)
+ .isThrownBy { WireFrame.decodeFirst(buff) }
+ }
+
+ it("should throw exception when payload doesn't fit") {
+ val buff = Unpooled.buffer()
+ .writeBytes(encodeSampleFrame())
+ buff.writerIndex(buff.writerIndex() - 2)
+
+ assertThatExceptionOfType(MissingWireFrameBytesException::class.java)
+ .isThrownBy { WireFrame.decodeFirst(buff) }
+ }
+
+ }
}
+
})
\ No newline at end of file
CMD ["--listen-port", "6061"]
COPY target/libs/external/* ./
COPY target/libs/internal/* ./
-COPY target/hv-collector-main-*.jar ./
\ No newline at end of file
+COPY target/hv-collector-main-*.jar ./
val collectorProvider = CollectorFactory(
resolveConfigurationProvider(serverConfiguration),
- AdapterFactory.kafkaSink()
+ AdapterFactory.loggingSink()
).createVesHvCollectorProvider()
ServerFactory.createNettyTcpServer(serverConfiguration, collectorProvider).start().block()
} catch (ex: WrongArgumentException) {
</rollingPolicy>
</appender>
- <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
+ <logger name="org.onap.dcae.collectors.veshv" level="INFO"/>
+ <!--<logger name="reactor.ipc.netty" level="DEBUG"/>-->
<root level="INFO">
<appender-ref ref="CONSOLE"/>
class Logger(val logger: org.slf4j.Logger) {
constructor(clazz: KClass<out Any>) : this(LoggerFactory.getLogger(clazz.java))
+ //
+ // TRACE
+ //
+
+ fun trace(messageProvider: () -> String) {
+ if (logger.isTraceEnabled) {
+ logger.trace(messageProvider())
+ }
+ }
+
+
//
// DEBUG
//
Code Review / dcaegen2 / collectors / hv-ves.git / commitdiff