sources/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/codec.kt

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * dcaegen2-collectors-veshv
   4  * ================================================================================
   5  * Copyright (C) 2018 NOKIA
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20 package org.onap.dcae.collectors.veshv.domain
  21
  22 import arrow.core.Either
  23 import arrow.core.Left
  24 import arrow.core.Right
  25 import io.netty.buffer.ByteBuf
  26 import io.netty.buffer.ByteBufAllocator
  27 import org.onap.dcae.collectors.veshv.domain.WireFrameMessage.Companion.RESERVED_BYTE_COUNT
  28
  29 /**
  30  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  31  * @since June 2018
  32  */
  33 class WireFrameEncoder(private val allocator: ByteBufAllocator = ByteBufAllocator.DEFAULT) {
  34
  35     fun encode(frame: WireFrameMessage): ByteBuf = allocator
  36             .buffer(WireFrameMessage.HEADER_SIZE + frame.payload.size())
  37             .run {
  38                 writeByte(WireFrameMessage.MARKER_BYTE.toInt())
  39                 writeByte(frame.versionMajor.toInt())
  40                 writeByte(frame.versionMinor.toInt())
  41                 writeZero(RESERVED_BYTE_COUNT)
  42                 writeShort(frame.payloadType)
  43                 writeInt(frame.payloadSize)
  44             }
  45             .also {
  46                 frame.payload.writeTo(it)
  47             }
  48 }
  49
  50 /**
  51  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  52  * @since June 2018
  53  */
  54 class WireFrameDecoder(private val maxPayloadSizeBytes: Int) {
  55
  56     fun decodeFirst(byteBuf: ByteBuf): Either<WireFrameDecodingError, WireFrameMessage> =
  57             when {
  58                 isEmpty(byteBuf) -> Left(EmptyWireFrame)
  59                 headerDoesNotFit(byteBuf) -> Left(MissingWireFrameHeaderBytes)
  60                 else -> parseWireFrame(byteBuf)
  61             }
  62
  63     private fun isEmpty(byteBuf: ByteBuf) = byteBuf.readableBytes() < 1
  64
  65     private fun headerDoesNotFit(byteBuf: ByteBuf) = byteBuf.readableBytes() < WireFrameMessage.HEADER_SIZE
  66
  67     private fun parseWireFrame(byteBuf: ByteBuf): Either<WireFrameDecodingError, WireFrameMessage> {
  68         byteBuf.markReaderIndex()
  69
  70         val mark = byteBuf.readUnsignedByte()
  71         return when (mark) {
  72             WireFrameMessage.MARKER_BYTE -> parsePayloadFrame(byteBuf)
  73             else -> {
  74                 byteBuf.resetReaderIndex()
  75                 Left(InvalidWireFrameMarker(mark))
  76             }
  77         }
  78     }
  79
  80     @Suppress("ReturnCount")
  81     private fun parsePayloadFrame(byteBuf: ByteBuf): Either<WireFrameDecodingError, WireFrameMessage> {
  82         val versionMajor = byteBuf.readUnsignedByte()
  83         val versionMinor = byteBuf.readUnsignedByte()
  84         byteBuf.skipBytes(RESERVED_BYTE_COUNT)
  85         val payloadTypeRaw = byteBuf.readUnsignedShort()
  86         val payloadSize = byteBuf.readInt()
  87
  88         if (payloadSize > maxPayloadSizeBytes) {
  89             byteBuf.resetReaderIndex()
  90             return Left(PayloadSizeExceeded(maxPayloadSizeBytes))
  91         }
  92
  93         if (byteBuf.readableBytes() < payloadSize) {
  94             byteBuf.resetReaderIndex()
  95             return Left(MissingWireFramePayloadBytes)
  96         }
  97
  98         val payload = ByteData.readFrom(byteBuf, payloadSize)
  99
 100         return Right(WireFrameMessage(payload, versionMajor, versionMinor, payloadTypeRaw, payloadSize))
 101     }
 102 }