Use Either instead of exceptions in frame decoder

Goals:
* Make code cleaner (in a FP way)
* Avoid costly exception throw each time we wait for the rest of the
frame (collecting stack traces is costly and we do not need them
anyway)

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

7 files changed, 68 insertions, 133 deletions

diff --git a/hv-collector-domain/pom.xml b/hv-collector-domain/pom.xml
index c11510ac..85c2a45a 100644
--- a/hv-collector-domain/pom.xml
+++ b/hv-collector-domain/pom.xml
@@ -103,6 +103,10 @@
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.arrow-kt</groupId>
+            <artifactId>arrow-core</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>org.assertj</groupId>
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 3cd9b19a..22767ed3 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
@@ -19,11 +19,11 @@
  */
 package org.onap.dcae.collectors.veshv.domain
 
+import arrow.core.Either
+import arrow.core.Left
+import arrow.core.Right
 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>
@@ -50,49 +50,37 @@ class WireFrameEncoder(val allocator: ByteBufAllocator) {
  */
 class WireFrameDecoder {
 
-    fun decodeFirst(byteBuf: ByteBuf): WireFrame {
-        verifyNotEmpty(byteBuf)
-        byteBuf.markReaderIndex()
-
-        verifyMarker(byteBuf)
-        verifyMinimumSize(byteBuf)
-
-        val version = byteBuf.readUnsignedByte()
-        val payloadTypeRaw = byteBuf.readUnsignedByte()
-        val payloadSize = verifyPayloadSize(byteBuf)
-        val payload = ByteData.readFrom(byteBuf, payloadSize)
+    fun decodeFirst(byteBuf: ByteBuf): Either<WireFrameDecodingError, WireFrame> =
+            when {
+                isEmpty(byteBuf)          -> Left(EmptyWireFrame)
+                headerDoesNotFit(byteBuf) -> Left(MissingWireFrameHeaderBytes)
+                else                      -> parseFrame(byteBuf)
+            }
 
-        return WireFrame(payload, version, payloadTypeRaw, payloadSize)
-    }
+    private fun headerDoesNotFit(byteBuf: ByteBuf) = byteBuf.readableBytes() < WireFrame.HEADER_SIZE
 
-    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 isEmpty(byteBuf: ByteBuf) = byteBuf.readableBytes() < 1
 
-    private fun verifyMinimumSize(byteBuf: ByteBuf) {
-        if (byteBuf.readableBytes() < WireFrame.HEADER_SIZE) {
-            byteBuf.resetReaderIndex()
-            throw MissingWireFrameBytesException("readable bytes < header size")
-        }
-    }
+    private fun parseFrame(byteBuf: ByteBuf): Either<WireFrameDecodingError, WireFrame> {
+        byteBuf.markReaderIndex()
 
-    private fun verifyMarker(byteBuf: ByteBuf) {
         val mark = byteBuf.readUnsignedByte()
         if (mark != WireFrame.MARKER_BYTE) {
             byteBuf.resetReaderIndex()
-            throw InvalidWireFrameMarkerException(mark)
+            return Left(InvalidWireFrameMarker(mark))
         }
-    }
 
-    private fun verifyNotEmpty(byteBuf: ByteBuf) {
-        if (byteBuf.readableBytes() < 1) {
-            throw EmptyWireFrameException()
+        val version = byteBuf.readUnsignedByte()
+        val payloadTypeRaw = byteBuf.readUnsignedByte()
+
+        val payloadSize = byteBuf.readInt()
+        if (byteBuf.readableBytes() < payloadSize) {
+            byteBuf.resetReaderIndex()
+            return Left(MissingWireFramePayloadBytes)
         }
+
+        val payload = ByteData.readFrom(byteBuf, payloadSize)
+
+        return Right(WireFrame(payload, version, payloadTypeRaw, payloadSize))
     }
 }
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/errors.kt
index ff452a7a..fb225202 100644
--- 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/errors.kt
@@ -17,13 +17,29 @@
  * limitations under the License.
  * ============LICENSE_END=========================================================
  */
-package org.onap.dcae.collectors.veshv.domain.exceptions
-
-import org.onap.dcae.collectors.veshv.domain.WireFrame
+package org.onap.dcae.collectors.veshv.domain
 
 /**
  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  * @since June 2018
  */
-class InvalidWireFrameMarkerException(actualMarker: Short) : WireFrameDecodingException(
+
+sealed class WireFrameDecodingError(val message: String)
+
+
+// Invalid frame errors
+
+sealed class InvalidWireFrame(msg: String) : WireFrameDecodingError(msg)
+
+class InvalidWireFrameMarker(actualMarker: Short)
+    : InvalidWireFrame(
         "Invalid start of frame. Expected 0x%02X, but was 0x%02X".format(WireFrame.MARKER_BYTE, actualMarker))
+
+
+// Missing bytes errors
+
+sealed class MissingWireFrameBytes(msg: String) : WireFrameDecodingError(msg)
+
+object MissingWireFrameHeaderBytes : MissingWireFrameBytes("readable bytes < header size")
+object MissingWireFramePayloadBytes : MissingWireFrameBytes("readable bytes < payload size")
+object EmptyWireFrame : MissingWireFrameBytes("empty wire frame")
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
deleted file mode 100644
index 6e1ce935..00000000
--- a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/EmptyWireFrameException.kt
+++ /dev/null
@@ -1,26 +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.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/MissingWireFrameBytesException.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/MissingWireFrameBytesException.kt
deleted file mode 100644
index 7e4b3cef..00000000
--- a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/MissingWireFrameBytesException.kt
+++ /dev/null
@@ -1,26 +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.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
deleted file mode 100644
index 11013834..00000000
--- a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/exceptions/WireFrameDecodingException.kt
+++ /dev/null
@@ -1,26 +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.exceptions
-
-/**
- * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
- * @since June 2018
- */
-open class WireFrameDecodingException(msg: String) : Exception(msg)
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 9694caf7..a97d889c 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
@@ -19,16 +19,17 @@
  */
 package org.onap.dcae.collectors.veshv.domain
 
+import arrow.core.Either
+import arrow.core.identity
 import io.netty.buffer.Unpooled
 import io.netty.buffer.UnpooledByteBufAllocator
 import org.assertj.core.api.Assertions.assertThat
-import org.assertj.core.api.Assertions.assertThatExceptionOfType
+import org.assertj.core.api.Assertions.fail
+import org.assertj.core.api.ObjectAssert
 import org.jetbrains.spek.api.Spek
 import org.jetbrains.spek.api.dsl.describe
 import org.jetbrains.spek.api.dsl.given
 import org.jetbrains.spek.api.dsl.it
-import org.onap.dcae.collectors.veshv.domain.exceptions.InvalidWireFrameMarkerException
-import org.onap.dcae.collectors.veshv.domain.exceptions.MissingWireFrameBytesException
 import java.nio.charset.Charset
 
 /**
@@ -119,7 +120,7 @@ object WireFrameCodecsTest : Spek({
         describe("encode-decode methods' compatibility") {
             val frame = createSampleFrame()
             val encoded = encodeSampleFrame()
-            val decoded = decoder.decodeFirst(encoded)
+            val decoded = decoder.decodeFirst(encoded).getOrFail()
 
             it("should decode version") {
                 assertThat(decoded.version).isEqualTo(frame.version)
@@ -146,7 +147,7 @@ object WireFrameCodecsTest : Spek({
                 val buff = Unpooled.buffer()
                         .writeBytes(encodeSampleFrame())
                         .writeByte(0xAA)
-                val decoded = decoder.decodeFirst(buff)
+                val decoded = decoder.decodeFirst(buff).getOrFail()
 
                 assertThat(decoded.isValid()).describedAs("should be valid").isTrue()
                 assertThat(buff.readableBytes()).isEqualTo(1)
@@ -156,8 +157,8 @@ object WireFrameCodecsTest : Spek({
                 val buff = Unpooled.buffer()
                         .writeByte(0xFF)
 
-                assertThatExceptionOfType(MissingWireFrameBytesException::class.java)
-                        .isThrownBy { decoder.decodeFirst(buff) }
+                decoder.decodeFirst(buff).assertFailedWithError { it.isInstanceOf(MissingWireFrameHeaderBytes::class.java) }
+
             }
 
             it("should throw exception when first byte is not 0xFF but length looks ok") {
@@ -165,16 +166,14 @@ object WireFrameCodecsTest : Spek({
                         .writeByte(0xAA)
                         .writeBytes("some garbage".toByteArray())
 
-                assertThatExceptionOfType(InvalidWireFrameMarkerException::class.java)
-                        .isThrownBy { decoder.decodeFirst(buff) }
+                decoder.decodeFirst(buff).assertFailedWithError { it.isInstanceOf(InvalidWireFrameMarker::class.java) }
             }
 
             it("should throw exception when first byte is not 0xFF and length is to short") {
                 val buff = Unpooled.buffer()
                         .writeByte(0xAA)
 
-                assertThatExceptionOfType(InvalidWireFrameMarkerException::class.java)
-                        .isThrownBy { decoder.decodeFirst(buff) }
+                decoder.decodeFirst(buff).assertFailedWithError { it.isInstanceOf(MissingWireFrameHeaderBytes::class.java) }
             }
 
             it("should throw exception when payload doesn't fit") {
@@ -182,11 +181,17 @@ object WireFrameCodecsTest : Spek({
                         .writeBytes(encodeSampleFrame())
                 buff.writerIndex(buff.writerIndex() - 2)
 
-                assertThatExceptionOfType(MissingWireFrameBytesException::class.java)
-                        .isThrownBy { decoder.decodeFirst(buff) }
+                decoder.decodeFirst(buff).assertFailedWithError { it.isInstanceOf(MissingWireFramePayloadBytes::class.java) }
             }
 
         }
     }
 
-})
\ No newline at end of file
+})
+
+private fun <A, B> Either<A, B>.assertFailedWithError(assertj: (ObjectAssert<A>) -> Unit) {
+    fold({ assertj(assertThat(it)) }, { fail("Error expected") })
+}
+
+private fun Either<WireFrameDecodingError, WireFrame>.getOrFail(): WireFrame =
+        fold({ fail(it.message) }, ::identity) as WireFrame