diff options
7 files changed, 144 insertions, 125 deletions
diff --git a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameMessages.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameMessages.kt deleted file mode 100644 index 4811a2b4..00000000 --- a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameMessages.kt +++ /dev/null @@ -1,107 +0,0 @@ -/* - * ============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 - - -sealed class WireFrameMessage - -/** - * Wire frame structure is presented bellow. All fields are in network byte order (big-endian). - * - * ``` - * ┌─────┬───────────────────────┬───────────────────────┬───────────────────────┐ - * │octet│ 0 │ 1 │ 2 │ - * ├─────┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┤ - * │ bit │ 0│ │ │ │ │ │ │ │ 8│ │ │ │ │ │ │ │16│ │ │ │ │ │ │ │ ... - * ├─────┼──┴──┴──┴──┴──┴──┴──┴──┼──┴──┴──┴──┴──┴──┴──┴──┼──┴──┴──┴──┴──┴──┴──┴──┤ - * │field│ 0xFF │ version │ payload content type │ - * └─────┴───────────────────────┴───────────────────────┴───────────────────────┘ - * ┌─────┬───────────────────────┬───────────────────────┬───────────────────────┬───────────────────────┐ - * │octet│ 3 │ 4 │ 5 │ 6 │ - * ├─────┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┤ - * ... │ bit │24│ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ... - * ├─────┼──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┤ - * │field│ payload size │ - * └─────┴───────────────────────────────────────────────────────────────────────────────────────────────┘ - * ┌─────┬─────────────────────── - * │octet│ 7 ... - * ├─────┼──┬──┬──┬──┬──┬──┬──┬── - * ... │ bit │56│ │ │ │ │ │ │... - * ├─────┼──┴──┴──┴──┴──┴──┴──┴── - * │field│ protobuf payload - * └─────┴─────────────────────── - * ``` - * - * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> - * @since May 2018 - */ -data class PayloadWireFrameMessage(val payload: ByteData, - val version: Short, - val payloadTypeRaw: Short, - val payloadSize: Int -) : WireFrameMessage() { - - constructor(payload: ByteArray) : this( - ByteData(payload), - SUPPORTED_VERSION, - PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue, - payload.size) - - fun isValid(): Boolean = - version == SUPPORTED_VERSION - && PayloadContentType.isValidHexValue(payloadTypeRaw) - && payload.size() == payloadSize - - companion object { - const val MARKER_BYTE: Short = 0xFF - - const val SUPPORTED_VERSION: Short = 1 - - const val HEADER_SIZE = - 3 * java.lang.Byte.BYTES + - 1 * java.lang.Integer.BYTES - - const val MAX_PAYLOAD_SIZE = 1024 * 1024 - } -} - - -/** - * This message type should be used by client to indicate that he has finished sending data to collector. - * - * Wire frame structure is presented bellow. All fields are in network byte order (big-endian). - * - * ``` - * ┌─────┬───────────────────────┐ - * │octet│ 0 │ - * ├─────┼──┬──┬──┬──┬──┬──┬──┬──┤ - * │ bit │ 0│ │ │ │ │ │ │ │ - * ├─────┼──┴──┴──┴──┴──┴──┴──┴──┤ - * │field│ 0xAA │ - * └─────┴───────────────────────┘ - * ``` - * - * @since July 2018 - */ - -object EndOfTransmissionMessage : WireFrameMessage() { - const val MARKER_BYTE: Short = 0xAA -} - diff --git a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/codec.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/codec.kt index cbc18fd0..b2e42509 100644 --- a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/codec.kt +++ b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/codec.kt @@ -25,6 +25,7 @@ import arrow.core.Right import io.netty.buffer.ByteBuf import io.netty.buffer.ByteBufAllocator import org.onap.dcae.collectors.veshv.domain.PayloadWireFrameMessage.Companion.MAX_PAYLOAD_SIZE +import org.onap.dcae.collectors.veshv.domain.PayloadWireFrameMessage.Companion.RESERVED_BYTE_COUNT /** * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> @@ -36,7 +37,9 @@ class WireFrameEncoder(private val allocator: ByteBufAllocator) { val bb = allocator.buffer(PayloadWireFrameMessage.HEADER_SIZE + frame.payload.size()) bb.writeByte(PayloadWireFrameMessage.MARKER_BYTE.toInt()) - bb.writeByte(frame.version.toInt()) + bb.writeByte(frame.versionMajor.toInt()) + bb.writeByte(frame.versionMinor.toInt()) + bb.writeZero(RESERVED_BYTE_COUNT) bb.writeByte(frame.payloadTypeRaw.toInt()) bb.writeInt(frame.payloadSize) frame.payload.writeTo(bb) @@ -92,9 +95,10 @@ class WireFrameDecoder { } private fun parsePayloadFrame(byteBuf: ByteBuf): Either<WireFrameDecodingError, PayloadWireFrameMessage> { - val version = byteBuf.readUnsignedByte() + val versionMajor = byteBuf.readUnsignedByte() + val versionMinor = byteBuf.readUnsignedByte() + byteBuf.skipBytes(RESERVED_BYTE_COUNT) // reserved val payloadTypeRaw = byteBuf.readUnsignedByte() - val payloadSize = byteBuf.readInt() if (payloadSize > MAX_PAYLOAD_SIZE) { @@ -109,7 +113,7 @@ class WireFrameDecoder { val payload = ByteData.readFrom(byteBuf, payloadSize) - return Right(PayloadWireFrameMessage(payload, version, payloadTypeRaw, payloadSize)) + return Right(PayloadWireFrameMessage(payload, versionMajor, versionMinor, payloadTypeRaw, payloadSize)) } } diff --git a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/wire_frame.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/wire_frame.kt new file mode 100644 index 00000000..642179e1 --- /dev/null +++ b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/wire_frame.kt @@ -0,0 +1,97 @@ +/* + * ============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 + + +sealed class WireFrameMessage + +/** + * Wire frame structure is presented bellow. All fields are in network byte order (big-endian). + * + * ``` + * -- Precedes every HV-VES message + * Header ::= SEQUENCE { + * magic INTEGER (0..255), – always 0xFF, identifies extended header usage + * versionMajor INTEGER (0..255), – major interface v, forward incompatible with previous major v + * versionMinor INTEGER (0..255), – minor interface v, forward compatible with previous minor v + * reserved BIT STRING (SIZE (16)), – reserved for future use + * messageType INTEGER (0..255), – message payload type: 0x00=undefined, 0x01=protobuf + * messageLength INTEGER (0..4294967295) – message payload length + * } + * ``` + * + * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> + * @since May 2018 + */ +data class PayloadWireFrameMessage(val payload: ByteData, + val versionMajor: Short, + val versionMinor: Short, + val payloadTypeRaw: Short, + val payloadSize: Int +) : WireFrameMessage() { + constructor(payload: ByteArray) : this( + ByteData(payload), + SUPPORTED_VERSION_MAJOR, + SUPPORTED_VERSION_MINOR, + PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue, + payload.size) + + fun isValid(): Boolean = + versionMajor == SUPPORTED_VERSION_MAJOR + && PayloadContentType.isValidHexValue(payloadTypeRaw) + && payload.size() == payloadSize + + companion object { + const val MARKER_BYTE: Short = 0xFF + const val RESERVED_BYTE_COUNT: Int = 3 + + const val SUPPORTED_VERSION_MAJOR: Short = 1 + const val SUPPORTED_VERSION_MINOR: Short = 0 + + const val HEADER_SIZE = + 1 * java.lang.Byte.BYTES + // marker + 3 * java.lang.Byte.BYTES + // single byte fields + RESERVED_BYTE_COUNT * java.lang.Byte.BYTES + // reserved bytes + 1 * java.lang.Integer.BYTES // payload length + + const val MAX_PAYLOAD_SIZE = 1024 * 1024 + } +} + + +/** + * This message type should be used by client to indicate that he has finished sending data to collector. + * + * Wire frame structure is presented bellow. All fields are in network byte order (big-endian). + * + * ``` + * -- Sent by the HV-VES data provider, prior to closing the connection to the HV-VES destination + * Eot ::= SEQUENCE { + * magic INTEGER (0..255), – always 0xAA + * } + * ``` + * + * @since July 2018 + */ + +object EndOfTransmissionMessage : WireFrameMessage() { + const val MARKER_BYTE: Short = 0xAA +} + diff --git a/hv-collector-domain/src/test/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameCodecsTest.kt b/hv-collector-domain/src/test/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameCodecsTest.kt index a1395266..89d1f32e 100644 --- a/hv-collector-domain/src/test/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameCodecsTest.kt +++ b/hv-collector-domain/src/test/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameCodecsTest.kt @@ -53,10 +53,11 @@ object WireFrameCodecsTest : Spek({ describe("Wire Frame invariants") { - given("input with unsupported version") { + given("input with unsupported major version") { val input = PayloadWireFrameMessage( payload = ByteData.EMPTY, - version = 100, + versionMajor = 100, + versionMinor = 0, payloadTypeRaw = PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue, payloadSize = 0) @@ -65,10 +66,24 @@ object WireFrameCodecsTest : Spek({ } } + given("input with unsupported minor version") { + val input = PayloadWireFrameMessage( + payload = ByteData.EMPTY, + versionMajor = 1, + versionMinor = 6, + payloadTypeRaw = PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue, + payloadSize = 0) + + it("should pass validation") { + assertThat(input.isValid()).isTrue() + } + } + given("input with unsupported payload type") { val input = PayloadWireFrameMessage( payload = ByteData.EMPTY, - version = 1, + versionMajor = 1, + versionMinor = 0, payloadTypeRaw = 0x69, payloadSize = 0) @@ -80,7 +95,8 @@ object WireFrameCodecsTest : Spek({ given("input with too small payload size") { val input = PayloadWireFrameMessage( payload = ByteData(byteArrayOf(1, 2, 3)), - version = 1, + versionMajor = 1, + versionMinor = 0, payloadTypeRaw = PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue, payloadSize = 1) @@ -92,7 +108,8 @@ object WireFrameCodecsTest : Spek({ given("input with too big payload size") { val input = PayloadWireFrameMessage( payload = ByteData(byteArrayOf(1, 2, 3)), - version = 1, + versionMajor = 1, + versionMinor = 0, payloadTypeRaw = PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue, payloadSize = 8) @@ -105,7 +122,8 @@ object WireFrameCodecsTest : Spek({ val payload = byteArrayOf(6, 9, 8, 6) val input = PayloadWireFrameMessage( payload = ByteData(payload), - version = 1, + versionMajor = 1, + versionMinor = 0, payloadTypeRaw = PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue, payloadSize = payload.size) @@ -125,7 +143,7 @@ object WireFrameCodecsTest : Spek({ val decoded = decoder.decodeFirst(encoded).getPayloadMessageOrFail() it("should decode version") { - assertThat(decoded.version).isEqualTo(frame.version) + assertThat(decoded.versionMajor).isEqualTo(frame.versionMajor) } it("should decode payload type") { @@ -221,7 +239,8 @@ object WireFrameCodecsTest : Spek({ val payload = ByteArray(MAX_PAYLOAD_SIZE) val input = PayloadWireFrameMessage( payload = ByteData(payload), - version = 1, + versionMajor = 1, + versionMinor = 0, payloadTypeRaw = PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue, payloadSize = payload.size) @@ -234,7 +253,8 @@ object WireFrameCodecsTest : Spek({ val payload = ByteArray(MAX_PAYLOAD_SIZE + 1) val input = PayloadWireFrameMessage( payload = ByteData(payload), - version = 1, + versionMajor = 1, + versionMinor = 0, payloadTypeRaw = PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue, payloadSize = payload.size) val buff = encoder.encode(input) @@ -249,7 +269,8 @@ object WireFrameCodecsTest : Spek({ val payload = ByteArray(MAX_PAYLOAD_SIZE) val input = PayloadWireFrameMessage( payload = ByteData(payload), - version = 1, + versionMajor = 1, + versionMinor = 0, payloadTypeRaw = PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue, payloadSize = payload.size) diff --git a/hv-collector-test-utils/src/main/kotlin/messages.kt b/hv-collector-test-utils/src/main/kotlin/messages.kt index f8453c64..c6aa89b2 100644 --- a/hv-collector-test-utils/src/main/kotlin/messages.kt +++ b/hv-collector-test-utils/src/main/kotlin/messages.kt @@ -24,6 +24,7 @@ import io.netty.buffer.ByteBuf import io.netty.buffer.ByteBufAllocator import io.netty.buffer.PooledByteBufAllocator import org.onap.dcae.collectors.veshv.domain.PayloadWireFrameMessage.Companion.MAX_PAYLOAD_SIZE +import org.onap.dcae.collectors.veshv.domain.PayloadWireFrameMessage.Companion.RESERVED_BYTE_COUNT import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain import java.util.UUID.randomUUID @@ -31,9 +32,11 @@ import java.util.UUID.randomUUID val allocator: ByteBufAllocator = PooledByteBufAllocator.DEFAULT private fun ByteBuf.writeValidWireFrameHeaders() { - writeByte(0xFF) // always 0xFF - writeByte(0x01) // version - writeByte(0x01) // content type = GPB + writeByte(0xFF) // always 0xFF + writeByte(0x01) // major version + writeByte(0x00) // minor version + writeZero(RESERVED_BYTE_COUNT) // reserved + writeByte(0x01) // content type = GPB } fun vesWireFrameMessage(domain: Domain = Domain.OTHER, diff --git a/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/MessageGeneratorImpl.kt b/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/MessageGeneratorImpl.kt index c6e0556b..fec2609e 100644 --- a/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/MessageGeneratorImpl.kt +++ b/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/MessageGeneratorImpl.kt @@ -71,6 +71,7 @@ class MessageGeneratorImpl internal constructor(private val payloadGenerator: Pa PayloadWireFrameMessage( payload, UNSUPPORTED_VERSION, + UNSUPPORTED_VERSION, PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue, payload.size()) } diff --git a/hv-collector-ves-message-generator/src/test/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/impl/MessageGeneratorImplTest.kt b/hv-collector-ves-message-generator/src/test/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/impl/MessageGeneratorImplTest.kt index 1b121ec4..f13a33bf 100644 --- a/hv-collector-ves-message-generator/src/test/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/impl/MessageGeneratorImplTest.kt +++ b/hv-collector-ves-message-generator/src/test/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/impl/MessageGeneratorImplTest.kt @@ -142,7 +142,7 @@ object MessageGeneratorImplTest : Spek({ assertThat(it.isValid()).isFalse() assertThat(it.payloadSize).isLessThan(PayloadWireFrameMessage.MAX_PAYLOAD_SIZE) assertThat(extractCommonEventHeader(it.payload).domain).isEqualTo(HVRANMEAS) - assertThat(it.version).isNotEqualTo(PayloadWireFrameMessage.SUPPORTED_VERSION) + assertThat(it.versionMajor).isNotEqualTo(PayloadWireFrameMessage.SUPPORTED_VERSION_MINOR) } .verifyComplete() } |