Fix TCP stream framing issue

Because of the nature of TCP protocol we receive consecutive IO buffer
snapshots - not separate messages. That means that we need to join
incomming buffers and then split it into separate WireFrames.

Closes ONAP-312
Change-Id: I84ba0ec58a41ff9026f2fca24d2b15f3adcf0a19
Signed-off-by: Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
Issue-ID: DCAEGEN2-601

5 files changed, 161 insertions, 13 deletions

diff --git a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt
index 5bd63d8b..8c8b4718 100644
--- a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt
+++ b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt
@@ -21,6 +21,9 @@ package org.onap.dcae.collectors.veshv.domain
 
 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).
@@ -53,20 +56,20 @@ import io.netty.buffer.ByteBufAllocator
  * @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)
@@ -76,20 +79,58 @@ data class WireFrame(val payload: ByteBuf,
     }
 
     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
     }
 }
diff --git a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/EmptyWireFrameException.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/EmptyWireFrameException.kt
new file mode 100644
index 00000000..6e1ce935
--- /dev/null
+++ b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/EmptyWireFrameException.kt
@@ -0,0 +1,26 @@
+/*
+ * ============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)")
diff --git a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/InvalidWireFrameMarkerException.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/InvalidWireFrameMarkerException.kt
new file mode 100644
index 00000000..ff452a7a
--- /dev/null
+++ b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/InvalidWireFrameMarkerException.kt
@@ -0,0 +1,29 @@
+/*
+ * ============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
+
+import org.onap.dcae.collectors.veshv.domain.WireFrame
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+class InvalidWireFrameMarkerException(actualMarker: Short) : WireFrameDecodingException(
+        "Invalid start of frame. Expected 0x%02X, but was 0x%02X".format(WireFrame.MARKER_BYTE, actualMarker))
diff --git a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/MissingWireFrameBytesException.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/MissingWireFrameBytesException.kt
new file mode 100644
index 00000000..7e4b3cef
--- /dev/null
+++ b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/MissingWireFrameBytesException.kt
@@ -0,0 +1,26 @@
+/*
+ * ============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)
diff --git a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/WireFrameDecodingException.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/WireFrameDecodingException.kt
new file mode 100644
index 00000000..11013834
--- /dev/null
+++ b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/WireFrameDecodingException.kt
@@ -0,0 +1,26 @@
+/*
+ * ============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)