Generate VesEvents in hv-ves/message-generator

- Split message generator on two specialized generators
  for VesEvent and WireFrame related message types
- Refactor whole message-generator module

Change-Id: I1266b549a9a4d27213d03e8921298deab2dacb59
Signed-off-by: Jakub Dudycz <jakub.dudycz@nokia.com>
Issue-ID: DCAEGEN2-1162

11 files changed, 272 insertions, 171 deletions

diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageGenerator.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageGenerator.kt
index 076c06be..5f8638f0 100644
--- a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageGenerator.kt
+++ b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageGenerator.kt
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * dcaegen2-collectors-veshv
  * ================================================================================
- * Copyright (C) 2018 NOKIA
+ * Copyright (C) 2018-2019 NOKIA
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,18 +19,26 @@
  */
 package org.onap.dcae.collectors.veshv.ves.message.generator.api
 
-import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
 import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
 
 /**
  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  * @since June 2018
  */
-interface MessageGenerator {
-    fun createMessageFlux(messageParameters: List<MessageParameters>): Flux<WireFrameMessage>
+abstract class MessageGenerator<K : MessageParameters, T> {
+    abstract fun createMessageFlux(parameters: K): Flux<T>
 
-    companion object {
-        const val FIXED_PAYLOAD_SIZE = 100
+    protected fun repeatMessage(message: Mono<T>, amount: Long): Flux<T> = when {
+        amount < 0 -> repeatForever(message)
+        amount == 0L -> emptyMessageStream()
+        else -> repeatNTimes(message, amount)
     }
+
+    private fun repeatForever(message: Mono<T>) = message.repeat()
+
+    private fun emptyMessageStream() = Flux.empty<T>()
+
+    private fun repeatNTimes(message: Mono<T>, amount: Long) = message.repeat(amount - 1)
 }
 
diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageParameters.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageParameters.kt
index 047d863c..82b79c0c 100644
--- a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageParameters.kt
+++ b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageParameters.kt
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * dcaegen2-collectors-veshv
  * ================================================================================
- * Copyright (C) 2018 NOKIA
+ * Copyright (C) 2018-2019 NOKIA
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,12 +19,25 @@
  */
 package org.onap.dcae.collectors.veshv.ves.message.generator.api
 
-import org.onap.ves.VesEventOuterClass.CommonEventHeader
+import org.onap.ves.VesEventOuterClass
 
 /**
  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
  * @since June 2018
  */
-data class MessageParameters(val commonEventHeader: CommonEventHeader,
-                             val messageType: MessageType,
-                             val amount: Long = -1)
+abstract class MessageParameters(val amount: Long = -1)
+
+/**
+ * @author Jakub Dudycz <jakub.dudycz@nokia.com>
+ * @since February 2019
+ */
+class WireFrameParameters(val messageType: WireFrameType,
+                          amount: Long = -1) : MessageParameters(amount)
+
+/**
+ * @author Jakub Dudycz <jakub.dudycz@nokia.com>
+ * @since February 2019
+ */
+class VesEventParameters(val commonEventHeader: VesEventOuterClass.CommonEventHeader,
+                         val messageType: VesEventType,
+                         amount: Long = -1) : MessageParameters(amount)
diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageParametersParser.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageParametersParser.kt
index 754fa31f..854c1cda 100644
--- a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageParametersParser.kt
+++ b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageParametersParser.kt
@@ -28,9 +28,7 @@ interface MessageParametersParser {
     fun parse(request: JsonArray): Either<ParsingError, List<MessageParameters>>
 
     companion object {
-        val INSTANCE: MessageParametersParser by lazy {
-            MessageParametersParserImpl()
-        }
+        val INSTANCE: MessageParametersParser by lazy { MessageParametersParserImpl() }
     }
 }
 
diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageType.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageTypes.kt
index 22c88252..54e26363 100644
--- a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageType.kt
+++ b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/api/MessageTypes.kt
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * dcaegen2-collectors-veshv
  * ================================================================================
- * Copyright (C) 2018 NOKIA
+ * Copyright (C) 2019 NOKIA
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,14 +19,31 @@
  */
 package org.onap.dcae.collectors.veshv.ves.message.generator.api
 
+import arrow.core.Try
+
 /**
  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
- * @since July 2018
+ * @since February 2019
  */
-enum class MessageType {
+enum class VesEventType {
     VALID,
     TOO_BIG_PAYLOAD,
-    FIXED_PAYLOAD,
+    FIXED_PAYLOAD;
+
+    companion object {
+        fun isVesEventType(str: String): Boolean = Try { valueOf(str) }.isSuccess()
+    }
+}
+
+/**
+ * @author Jakub Dudycz <jakub.dudycz@nokia.com>
+ * @since February 2019
+ */
+enum class WireFrameType {
     INVALID_WIRE_FRAME,
-    INVALID_GPB_DATA,
+    INVALID_GPB_DATA;
+
+    companion object {
+        fun isWireFrameType(str: String): Boolean = Try { WireFrameType.valueOf(str) }.isSuccess()
+    }
 }
diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/factory/MessageGeneratorFactory.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/factory/MessageGeneratorFactory.kt
index e2269c20..aa473796 100644
--- a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/factory/MessageGeneratorFactory.kt
+++ b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/factory/MessageGeneratorFactory.kt
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * dcaegen2-collectors-veshv
  * ================================================================================
- * Copyright (C) 2018 NOKIA
+ * Copyright (C) 2018-2019 NOKIA
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,15 +19,16 @@
  */
 package org.onap.dcae.collectors.veshv.ves.message.generator.factory
 
-import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageGenerator
-import org.onap.dcae.collectors.veshv.ves.message.generator.impl.MessageGeneratorImpl
-import org.onap.dcae.collectors.veshv.ves.message.generator.impl.PayloadGenerator
+import org.onap.dcae.collectors.veshv.ves.message.generator.impl.vesevent.PayloadGenerator
+import org.onap.dcae.collectors.veshv.ves.message.generator.impl.vesevent.VesEventGenerator
+import org.onap.dcae.collectors.veshv.ves.message.generator.impl.wireframe.WireFrameGenerator
 
 /**
  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  * @since October 2018
  */
-object MessageGeneratorFactory {
-    fun create(maxPayloadSizeBytes: Int): MessageGenerator =
-            MessageGeneratorImpl(PayloadGenerator(), maxPayloadSizeBytes)
+class MessageGeneratorFactory(private val maxPayloadSizeBytes: Int) {
+    fun createVesEventGenerator() = VesEventGenerator(PayloadGenerator(), maxPayloadSizeBytes)
+
+    fun createWireFrameGenerator() = WireFrameGenerator()
 }
diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/MessageGeneratorImpl.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/MessageGeneratorImpl.kt
deleted file mode 100644
index fa39ed16..00000000
--- a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/MessageGeneratorImpl.kt
+++ /dev/null
@@ -1,109 +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.ves.message.generator.impl
-
-import com.google.protobuf.ByteString
-import org.onap.dcae.collectors.veshv.domain.ByteData
-import org.onap.dcae.collectors.veshv.domain.PayloadContentType
-import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
-import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageGenerator
-import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParameters
-import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageType
-import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageType.FIXED_PAYLOAD
-import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageType.INVALID_GPB_DATA
-import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageType.INVALID_WIRE_FRAME
-import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageType.TOO_BIG_PAYLOAD
-import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageType.VALID
-import org.onap.ves.VesEventOuterClass.CommonEventHeader
-import org.onap.ves.VesEventOuterClass.VesEvent
-import reactor.core.publisher.Flux
-import reactor.core.publisher.Mono
-import java.nio.charset.Charset
-
-/**
- * @author Jakub Dudycz <jakub.dudycz@nokia.com>
- * @since June 2018
- */
-class MessageGeneratorImpl internal constructor(
-        private val payloadGenerator: PayloadGenerator,
-        private val maxPayloadSizeBytes: Int
-) : MessageGenerator {
-
-    override fun createMessageFlux(messageParameters: List<MessageParameters>): Flux<WireFrameMessage> = Flux
-            .fromIterable(messageParameters)
-            .flatMap { createMessageFlux(it) }
-
-    private fun createMessageFlux(parameters: MessageParameters): Flux<WireFrameMessage> =
-            Mono.fromCallable { createMessage(parameters.commonEventHeader, parameters.messageType) }
-                    .let {
-                        when {
-                            parameters.amount < 0 ->
-                                // repeat forever
-                                it.repeat()
-                            parameters.amount == 0L ->
-                                // do not generate any message
-                                Flux.empty()
-                            else ->
-                                // send original message and additional amount-1 messages
-                                it.repeat(parameters.amount - 1)
-                        }
-                    }
-
-    private fun createMessage(commonEventHeader: CommonEventHeader, messageType: MessageType): WireFrameMessage =
-            when (messageType) {
-                VALID ->
-                    WireFrameMessage(vesEvent(commonEventHeader, payloadGenerator.generatePayload()))
-                TOO_BIG_PAYLOAD ->
-                    WireFrameMessage(vesEvent(commonEventHeader, oversizedPayload()))
-                FIXED_PAYLOAD ->
-                    WireFrameMessage(vesEvent(commonEventHeader, fixedPayload()))
-                INVALID_WIRE_FRAME -> {
-                    val payload = ByteData(vesEvent(commonEventHeader, payloadGenerator.generatePayload()))
-                    WireFrameMessage(
-                            payload,
-                            UNSUPPORTED_VERSION,
-                            UNSUPPORTED_VERSION,
-                            PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue,
-                            payload.size())
-                }
-                INVALID_GPB_DATA ->
-                    WireFrameMessage("invalid vesEvent".toByteArray(Charset.defaultCharset()))
-            }
-
-    private fun vesEvent(commonEventHeader: CommonEventHeader, eventFields: ByteString): ByteArray {
-        return createVesEvent(commonEventHeader, eventFields).toByteArray()
-    }
-
-    private fun createVesEvent(commonEventHeader: CommonEventHeader, payload: ByteString): VesEvent =
-            VesEvent.newBuilder()
-                    .setCommonEventHeader(commonEventHeader)
-                    .setEventFields(payload)
-                    .build()
-
-    private fun oversizedPayload() =
-            payloadGenerator.generateRawPayload(maxPayloadSizeBytes + 1)
-
-    private fun fixedPayload() =
-            payloadGenerator.generateRawPayload(MessageGenerator.FIXED_PAYLOAD_SIZE)
-
-    companion object {
-        private const val UNSUPPORTED_VERSION: Short = 2
-    }
-}
diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/MessageParametersParserImpl.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/MessageParametersParserImpl.kt
index 88cc47a6..174a01fd 100644
--- a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/MessageParametersParserImpl.kt
+++ b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/MessageParametersParserImpl.kt
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * dcaegen2-collectors-veshv
  * ================================================================================
- * Copyright (C) 2018 NOKIA
+ * Copyright (C) 2018-2019 NOKIA
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,15 +19,24 @@
  */
 package org.onap.dcae.collectors.veshv.ves.message.generator.impl
 
+import arrow.core.Either
 import arrow.core.Option
 import arrow.core.Try
 import arrow.core.identity
 import org.onap.dcae.collectors.veshv.utils.logging.Logger
 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParameters
 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParametersParser
-import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageType
 import org.onap.dcae.collectors.veshv.ves.message.generator.api.ParsingError
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.VesEventParameters
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.VesEventType
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.VesEventType.Companion.isVesEventType
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.WireFrameParameters
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.WireFrameType
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.WireFrameType.Companion.isWireFrameType
+import org.onap.dcae.collectors.veshv.ves.message.generator.impl.vesevent.CommonEventHeaderParser
 import javax.json.JsonArray
+import javax.json.JsonObject
+import javax.json.JsonValue
 
 /**
  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
@@ -37,28 +46,49 @@ internal class MessageParametersParserImpl(
         private val commonEventHeaderParser: CommonEventHeaderParser = CommonEventHeaderParser()
 ) : MessageParametersParser {
 
-    override fun parse(request: JsonArray) =
-            Try {
-                request
-                        .map { it.asJsonObject() }
-                        .onEach { logger.info { "Parsing MessageParameters body: $it" } }
-                        .map { json ->
-                            val commonEventHeader = commonEventHeaderParser
-                                    .parse(json.getJsonObject("commonEventHeader"))
-                                    .fold({ throw IllegalStateException("Invalid common header") }, ::identity)
-                            val messageType = MessageType.valueOf(json.getString("messageType"))
-                            val messagesAmount = json.getJsonNumber("messagesAmount")?.longValue()
-                                    ?: throw NullPointerException("\"messagesAmount\" could not be parsed.")
-                            MessageParameters(commonEventHeader, messageType, messagesAmount)
-                        }
-            }.toEither().mapLeft { ex ->
-                ParsingError(
-                        ex.message ?: "Unable to parse message parameters",
-                        Option.fromNullable(ex))
-            }
+    override fun parse(request: JsonArray): Either<ParsingError, List<MessageParameters>> =
+            Try { parseArray(request) }
+                    .toEither()
+                    .mapLeft { ex ->
+                        ParsingError(
+                                ex.message ?: "Unable to parse message parameters",
+                                Option.fromNullable(ex))
+                    }
+
+    private fun parseArray(array: JsonArray) = array
+            .map(JsonValue::asJsonObject)
+            .onEach { logger.info { "Parsing MessageParameters body: $it" } }
+            .map(::parseParameters)
+
+    private fun parseParameters(json: JsonObject): MessageParameters {
+        val messagesAmount = json.getJsonNumber("messagesAmount")?.longValue()
+                ?: throw ParsingException("\"messagesAmount\" could not be parsed.")
+
+        val messageType = json.getString("messageType")
+
+        return when {
+            isVesEventType(messageType) ->
+                constructVesEventParams(json, messageType, messagesAmount)
+            isWireFrameType(messageType) ->
+                WireFrameParameters(WireFrameType.valueOf(messageType), messagesAmount)
+            else -> throw ParsingException("Invalid message type")
+        }
+    }
+
+    private fun constructVesEventParams(json: JsonObject,
+                                        messageType: String,
+                                        messagesAmount: Long): VesEventParameters =
+            commonEventHeaderParser
+                    .parse(json.getJsonObject("commonEventHeader"))
+                    .fold({ throw ParsingException("Invalid common header") }, ::identity)
+                    .let { VesEventParameters(it, VesEventType.valueOf(messageType), messagesAmount) }
+
+
+    private class ParsingException(message: String) : Exception(message)
 
     companion object {
         private val logger = Logger(MessageParametersParserImpl::class)
     }
-
 }
+
+
diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/CommonEventHeaderParser.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/vesevent/CommonEventHeaderParser.kt
index 909db5e4..05938924 100644
--- a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/CommonEventHeaderParser.kt
+++ b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/vesevent/CommonEventHeaderParser.kt
@@ -17,7 +17,7 @@
  * limitations under the License.
  * ============LICENSE_END=========================================================
  */
-package org.onap.dcae.collectors.veshv.ves.message.generator.impl
+package org.onap.dcae.collectors.veshv.ves.message.generator.impl.vesevent
 
 import arrow.core.Option
 import com.google.protobuf.util.JsonFormat
@@ -30,18 +30,17 @@ import javax.json.JsonObject
  * @since July 2018
  */
 class CommonEventHeaderParser {
-    fun parse(json: JsonObject): Option<CommonEventHeader> =
-            Option.fromNullable(
-                    CommonEventHeader.newBuilder()
-                            .apply { JsonFormat.parser().merge(json.toString(), this) }
-                            .build()
-                            .takeUnless { !isValid(it) }
-            )
+    fun parse(json: JsonObject): Option<CommonEventHeader> = Option.fromNullable(
+            CommonEventHeader.newBuilder()
+                    .apply { JsonFormat.parser().merge(json.toString(), this) }
+                    .build()
+                    .takeUnless { !isValid(it) }
+    )
 
 
-    private fun isValid(header: CommonEventHeader): Boolean {
-        return allMandatoryFieldsArePresent(header)
-    }
+    private fun isValid(header: CommonEventHeader): Boolean =
+            allMandatoryFieldsArePresent(header)
+
 
     private fun allMandatoryFieldsArePresent(header: CommonEventHeader) =
             headerRequiredFieldDescriptors
diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/PayloadGenerator.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/vesevent/PayloadGenerator.kt
index 545e237c..ed521054 100644
--- a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/PayloadGenerator.kt
+++ b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/vesevent/PayloadGenerator.kt
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * dcaegen2-collectors-veshv
  * ================================================================================
- * Copyright (C) 2018 NOKIA
+ * Copyright (C) 2018-2019 NOKIA
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,7 +17,7 @@
  * limitations under the License.
  * ============LICENSE_END=========================================================
  */
-package org.onap.dcae.collectors.veshv.ves.message.generator.impl
+package org.onap.dcae.collectors.veshv.ves.message.generator.impl.vesevent
 
 import com.google.protobuf.ByteString
 import java.util.*
@@ -32,9 +32,15 @@ internal class PayloadGenerator {
 
     fun generatePayload(numOfCountMeasurements: Long = 2): ByteString =
             ByteString.copyFrom(
-                    randomGenerator.ints(numOfCountMeasurements, 0, 256)
+                    randomGenerator
+                            .ints(numOfCountMeasurements, MIN_BYTE_VALUE, MAX_BYTE_VALUE)
                             .asSequence()
                             .toString()
                             .toByteArray()
             )
+
+    companion object {
+        private const val MIN_BYTE_VALUE = 0
+        private const val MAX_BYTE_VALUE = 256
+    }
 }
diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/vesevent/VesEventGenerator.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/vesevent/VesEventGenerator.kt
new file mode 100644
index 00000000..7abd6054
--- /dev/null
+++ b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/vesevent/VesEventGenerator.kt
@@ -0,0 +1,72 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018-2019 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.ves.message.generator.impl.vesevent
+
+import com.google.protobuf.ByteString
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageGenerator
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.VesEventParameters
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.VesEventType
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.VesEventType.FIXED_PAYLOAD
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.VesEventType.TOO_BIG_PAYLOAD
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.VesEventType.VALID
+import org.onap.ves.VesEventOuterClass.CommonEventHeader
+import org.onap.ves.VesEventOuterClass.VesEvent
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+/**
+ * @author Jakub Dudycz <jakub.dudycz@nokia.com>
+ * @since June 2018
+ */
+class VesEventGenerator internal constructor(
+        private val payloadGenerator: PayloadGenerator,
+        private val maxPayloadSizeBytes: Int
+) : MessageGenerator<VesEventParameters, VesEvent>() {
+
+    override fun createMessageFlux(parameters: VesEventParameters): Flux<VesEvent> =
+            parameters.run {
+                Mono
+                        .fromCallable { createMessage(commonEventHeader, messageType) }
+                        .let { repeatMessage(it, amount) }
+            }
+
+    private fun createMessage(commonEventHeader: CommonEventHeader, messageType: VesEventType): VesEvent =
+            when (messageType) {
+                VALID -> vesEvent(commonEventHeader, payloadGenerator.generatePayload())
+                TOO_BIG_PAYLOAD -> vesEvent(commonEventHeader, oversizedPayload())
+                FIXED_PAYLOAD -> vesEvent(commonEventHeader, fixedPayload())
+            }
+
+    private fun vesEvent(commonEventHeader: CommonEventHeader, payload: ByteString): VesEvent =
+            VesEvent.newBuilder()
+                    .setCommonEventHeader(commonEventHeader)
+                    .setEventFields(payload)
+                    .build()
+
+    private fun oversizedPayload(): ByteString =
+            payloadGenerator.generateRawPayload(maxPayloadSizeBytes + 1)
+
+    private fun fixedPayload(): ByteString =
+            payloadGenerator.generateRawPayload(FIXED_PAYLOAD_SIZE)
+
+    companion object {
+        const val FIXED_PAYLOAD_SIZE = 100
+    }
+}
diff --git a/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/wireframe/WireFrameGenerator.kt b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/wireframe/WireFrameGenerator.kt
new file mode 100644
index 00000000..ad45bc5c
--- /dev/null
+++ b/sources/hv-collector-ves-message-generator/src/main/kotlin/org/onap/dcae/collectors/veshv/ves/message/generator/impl/wireframe/WireFrameGenerator.kt
@@ -0,0 +1,66 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2019 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.ves.message.generator.impl.wireframe
+
+import org.onap.dcae.collectors.veshv.domain.ByteData
+import org.onap.dcae.collectors.veshv.domain.PayloadContentType
+import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageGenerator
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.WireFrameParameters
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.WireFrameType
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.WireFrameType.INVALID_GPB_DATA
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.WireFrameType.INVALID_WIRE_FRAME
+import org.onap.ves.VesEventOuterClass.VesEvent
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import java.nio.charset.Charset
+
+/**
+ * @author Jakub Dudycz <jakub.dudycz@nokia.com>
+ * @since February 2019
+ */
+class WireFrameGenerator : MessageGenerator<WireFrameParameters, WireFrameMessage>() {
+
+    override fun createMessageFlux(parameters: WireFrameParameters): Flux<WireFrameMessage> =
+            parameters.run {
+                Mono
+                        .fromCallable { createMessage(messageType) }
+                        .let { repeatMessage(it, amount) }
+            }
+
+    private fun createMessage(messageType: WireFrameType): WireFrameMessage =
+            when (messageType) {
+                INVALID_WIRE_FRAME -> {
+                    val payload = ByteData(VesEvent.getDefaultInstance().toByteArray())
+                    WireFrameMessage(
+                            payload,
+                            UNSUPPORTED_VERSION,
+                            UNSUPPORTED_VERSION,
+                            PayloadContentType.GOOGLE_PROTOCOL_BUFFER.hexValue,
+                            payload.size())
+                }
+                INVALID_GPB_DATA ->
+                    WireFrameMessage("invalid vesEvent".toByteArray(Charset.defaultCharset()))
+            }
+
+    companion object {
+        private const val UNSUPPORTED_VERSION: Short = 2
+    }
+}