Implement Kafka Sink

Closes ONAP-146

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

3 files changed, 162 insertions, 69 deletions

diff --git a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/ByteData.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/ByteData.kt
new file mode 100644
index 00000000..2b84e3f1
--- /dev/null
+++ b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/ByteData.kt
@@ -0,0 +1,58 @@
+/*
+ * ============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 com.google.protobuf.MessageLite
+import io.netty.buffer.ByteBuf
+import java.nio.charset.Charset
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+class ByteData(private val data: ByteArray) {
+
+    fun size() = data.size
+
+    /**
+     * This will expose mutable state of the data.
+     *
+     * @return wrapped data buffer (NOT a copy)
+     */
+    fun unsafeAsArray() = data
+
+    fun writeTo(byteBuf: ByteBuf) {
+        byteBuf.writeBytes(data)
+    }
+
+    fun asString(charset: Charset = Charset.defaultCharset()) = String(data, charset)
+
+    companion object {
+        val EMPTY = ByteData(byteArrayOf())
+
+        fun readFrom(byteBuf: ByteBuf, length: Int): ByteData {
+            val dataArray = ByteArray(length)
+            byteBuf.readBytes(dataArray)
+            return ByteData(dataArray)
+        }
+    }
+}
+
+fun MessageLite.toByteData(): ByteData = ByteData(toByteArray())
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 8c8b4718..caf13c53 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
@@ -19,12 +19,6 @@
  */
 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).
  *
@@ -55,82 +49,25 @@ import org.onap.dcae.collectors.veshv.domain.exceptions.MissingWireFrameBytesExc
  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  * @since May 2018
  */
-data class WireFrame(val payload: ByteBuf,
+data class WireFrame(val payload: ByteData,
                      val majorVersion: Short,
                      val minorVersion: Short,
                      val payloadSize: Int) {
 
-    constructor(payload: ByteBuf) : this(payload, 1, 0, payload.readableBytes())
+    constructor(payload: ByteArray) : this(ByteData(payload), 1, 0, payload.size)
 
     fun isValid(): Boolean =
             majorVersion == SUPPORTED_MAJOR_VERSION
-                    && payload.readableBytes() == payloadSize
-
-    fun encode(allocator: ByteBufAllocator): ByteBuf {
-        val bb = allocator.buffer(HEADER_SIZE + payload.readableBytes())
-
-        bb.writeByte(MARKER_BYTE.toInt())
-        bb.writeByte(majorVersion.toInt())
-        bb.writeByte(minorVersion.toInt())
-        bb.writeInt(payloadSize)
-        bb.writeBytes(payload)
-
-        return bb
-    }
+                    && payload.size() == payloadSize
 
     companion object {
-        fun decodeFirst(byteBuf: ByteBuf): WireFrame {
-            verifyNotEmpty(byteBuf)
-            byteBuf.markReaderIndex()
-
-            verifyMarker(byteBuf)
-            verifyMinimumSize(byteBuf)
-
-            val majorVersion = byteBuf.readUnsignedByte()
-            val minorVersion = byteBuf.readUnsignedByte()
-            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)
-            }
-        }
-
-        private fun verifyNotEmpty(byteBuf: ByteBuf) {
-            if (byteBuf.readableBytes() < 1) {
-                throw EmptyWireFrameException()
-            }
-        }
+        const val SUPPORTED_MAJOR_VERSION: Short = 1
 
         const val HEADER_SIZE =
                 3 * java.lang.Byte.BYTES +
                         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/codec.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/codec.kt
new file mode 100644
index 00000000..d6804c7d
--- /dev/null
+++ b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/codec.kt
@@ -0,0 +1,98 @@
+/*
+ * ============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.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
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+class WireFrameEncoder(val allocator: ByteBufAllocator) {
+
+    fun encode(frame: WireFrame): ByteBuf {
+        val bb = allocator.buffer(WireFrame.HEADER_SIZE + frame.payload.size())
+
+        bb.writeByte(WireFrame.MARKER_BYTE.toInt())
+        bb.writeByte(frame.majorVersion.toInt())
+        bb.writeByte(frame.minorVersion.toInt())
+        bb.writeInt(frame.payloadSize)
+        frame.payload.writeTo(bb)
+
+        return bb
+    }
+}
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since June 2018
+ */
+class WireFrameDecoder {
+
+    fun decodeFirst(byteBuf: ByteBuf): WireFrame {
+        verifyNotEmpty(byteBuf)
+        byteBuf.markReaderIndex()
+
+        verifyMarker(byteBuf)
+        verifyMinimumSize(byteBuf)
+
+        val majorVersion = byteBuf.readUnsignedByte()
+        val minorVersion = byteBuf.readUnsignedByte()
+        val payloadSize = verifyPayloadSize(byteBuf)
+        val payload = ByteData.readFrom(byteBuf, 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() < WireFrame.HEADER_SIZE) {
+            byteBuf.resetReaderIndex()
+            throw MissingWireFrameBytesException("readable bytes < header size")
+        }
+    }
+
+    private fun verifyMarker(byteBuf: ByteBuf) {
+        val mark = byteBuf.readUnsignedByte()
+        if (mark != WireFrame.MARKER_BYTE) {
+            byteBuf.resetReaderIndex()
+            throw InvalidWireFrameMarkerException(mark)
+        }
+    }
+
+    private fun verifyNotEmpty(byteBuf: ByteBuf) {
+        if (byteBuf.readableBytes() < 1) {
+            throw EmptyWireFrameException()
+        }
+    }
+}