diff options
4 files changed, 72 insertions, 11 deletions
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt index 306b7762..ffd59bdc 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt @@ -81,7 +81,7 @@ data class WireFrame(val payload: ByteBuf, val majorVersion = byteBuf.readUnsignedByte() val minorVersion = byteBuf.readUnsignedByte() val payloadSize = byteBuf.readInt() - val payload = byteBuf.slice() + val payload = byteBuf.retainedSlice() return WireFrame(payload, mark, majorVersion, minorVersion, payloadSize) } 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 af9d0b0a..a3f26ce5 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 @@ -32,11 +32,11 @@ import reactor.core.publisher.Mono * @since May 2018 */ internal class VesHvCollector( - val wireDecoder: WireDecoder, - val protobufDecoder: VesDecoder, - val validator: MessageValidator, - val router: Router, - val sink: Sink) : Collector { + private val wireDecoder: 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 .flatMap(this::decodeWire) @@ -47,7 +47,7 @@ internal class VesHvCollector( .doOnNext(this::releaseMemory) .then() - private fun decodeWire(wire: ByteBuf) = releaseWhenNull(wire, wireDecoder::decode) + private fun decodeWire(wire: ByteBuf) = omitWhenNull(wire, wireDecoder::decode) private fun decodeProtobuf(protobuf: ByteBuf) = releaseWhenNull(protobuf, protobufDecoder::decode) @@ -71,6 +71,11 @@ internal class VesHvCollector( msg.rawMessage.release() } + + + private fun <T>omitWhenNull(input: ByteBuf, mapper: (ByteBuf) -> T?): Mono<T> = + Mono.justOrEmpty(mapper(input)) + private fun <T>releaseWhenNull(input: ByteBuf, mapper: (ByteBuf) -> T?): Mono<T> { val result = mapper(input) return if (result == null) { diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameTest.kt new file mode 100644 index 00000000..5a923c4e --- /dev/null +++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameTest.kt @@ -0,0 +1,55 @@ +package org.onap.dcae.collectors.veshv.domain + +import io.netty.buffer.ByteBufAllocator +import io.netty.buffer.Unpooled +import org.assertj.core.api.Assertions.assertThat +import org.jetbrains.spek.api.Spek +import org.jetbrains.spek.api.dsl.describe +import org.jetbrains.spek.api.dsl.it + +/** + * @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, + minorVersion = 2, + mark = 0xFF, + payloadSize = payload.readableBytes()) + + val encoded = frame.encode(ByteBufAllocator.DEFAULT) + val decoded = WireFrame.decode(encoded) + + it("should decode major version") { + assertThat(decoded.majorVersion).isEqualTo(frame.majorVersion) + } + + it("should decode minor version") { + 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) + } + + it("should retain decoded payload") { + encoded.release() + assertThat(decoded.payload.refCnt()).isEqualTo(1) + } + } + } +})
\ No newline at end of file diff --git a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt index 2cfb785e..5990fd0a 100644 --- a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt +++ b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt @@ -47,21 +47,22 @@ object VesHvSpecification : Spek({ val msgWithInvalidPayload = invalidVesMessage() val msgWithInvalidFrame = invalidWireFrame() val validMessage = vesMessage(Domain.HVRANMEAS) + val refCntBeforeSending = msgWithInvalidDomain.refCnt() sut.handleConnection(msgWithInvalidDomain, msgWithInvalidPayload, msgWithInvalidFrame, validMessage) assertThat(msgWithInvalidDomain.refCnt()) .describedAs("message with invalid domain should be released") - .isEqualTo(0) + .isEqualTo(refCntBeforeSending) assertThat(msgWithInvalidPayload.refCnt()) .describedAs("message with invalid payload should be released") - .isEqualTo(0) + .isEqualTo(refCntBeforeSending) assertThat(msgWithInvalidFrame.refCnt()) .describedAs("message with invalid frame should be released") - .isEqualTo(0) + .isEqualTo(refCntBeforeSending) assertThat(validMessage.refCnt()) .describedAs("handled message should be released") - .isEqualTo(0) + .isEqualTo(refCntBeforeSending) } } |