gerrit.onap Code Review - dcaegen2/collectors/hv-ves.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * dcaegen2-collectors-veshv
   4  * ================================================================================
   5  * Copyright (C) 2018-2019 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.simulators.dcaeapp.impl
  21
  22 import org.onap.dcae.collectors.veshv.utils.logging.Logger
  23 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParameters
  24 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParametersParser
  25 import org.onap.dcae.collectors.veshv.ves.message.generator.api.VesEventParameters
  26 import org.onap.dcae.collectors.veshv.ves.message.generator.api.VesEventType.FIXED_PAYLOAD
  27 import org.onap.dcae.collectors.veshv.ves.message.generator.generators.VesEventGenerator
  28 import org.onap.ves.VesEventOuterClass.VesEvent
  29 import reactor.core.publisher.Flux
  30 import reactor.core.publisher.Mono
  31 import java.io.InputStream
  32 import javax.json.Json
  33
  34 class MessageStreamValidation(
  35         private val messageGenerator: VesEventGenerator,
  36         private val messageParametersParser: MessageParametersParser = MessageParametersParser.INSTANCE) {
  37
  38     fun validate(jsonDescription: InputStream, consumedMessages: List<ByteArray>) =
  39             Mono
  40                     .fromSupplier { parseMessageParams(jsonDescription) }
  41                     .doOnNext {
  42                         logger.debug { "Parsed message parameters: $it" }
  43                     }
  44                     .flatMap { messageParams ->
  45                         val actualEvents = decodeConsumedEvents(consumedMessages)
  46                         generateEvents(messageParams).map {
  47                             if (shouldValidatePayloads(messageParams))
  48                                 it == actualEvents
  49                             else
  50                                 validateHeaders(actualEvents, it)
  51                         }
  52                     }
  53
  54     private fun parseMessageParams(input: InputStream): List<VesEventParameters> {
  55         val paramsArray = Json.createReader(input).readArray()
  56         val messageParams = messageParametersParser.parse(paramsArray)
  57
  58         return messageParams.fold(
  59                 {
  60                     logger.warn { "Error while parsing message parameters: ${it::class.qualifiedName} : ${it.message}" }
  61                     logger.debug { "Detailed stack trace: $it" }
  62                     throw IllegalArgumentException("Parsing error: " + it.message)
  63                 },
  64                 {
  65                     toVesEventParams(it)
  66                 }
  67         )
  68     }
  69
  70     private fun toVesEventParams(params: List<MessageParameters>): List<VesEventParameters> =
  71             if (params.isEmpty()) {
  72                 val message = "Message param list cannot be empty"
  73                 logger.warn { message }
  74                 throw IllegalArgumentException(message)
  75             } else params.map(::validateMessageParams)
  76
  77
  78     private fun validateMessageParams(params: MessageParameters): VesEventParameters =
  79             if (params !is VesEventParameters) {
  80                 val message = "Only VesEvent-related message types can be validated. " +
  81                         "Correct values are: VALID, TOO_BIG_PAYLOAD, FIXED_PAYLOAD"
  82                 logger.warn { message }
  83                 throw IllegalArgumentException(message)
  84             } else params
  85
  86
  87     private fun shouldValidatePayloads(parameters: List<VesEventParameters>) =
  88             parameters.all { it.messageType == FIXED_PAYLOAD }
  89
  90     private fun validateHeaders(actual: List<VesEvent>,
  91                                 expected: List<VesEvent>): Boolean {
  92         val consumedHeaders = actual.map { it.commonEventHeader }
  93         val generatedHeaders = expected.map { it.commonEventHeader }
  94         return generatedHeaders == consumedHeaders
  95     }
  96
  97     private fun generateEvents(parameters: List<VesEventParameters>): Mono<List<VesEvent>> = Flux
  98             .fromIterable(parameters)
  99             .flatMap { messageGenerator.createMessageFlux(it) }
 100             .collectList()
 101
 102     private fun decodeConsumedEvents(consumedMessages: List<ByteArray>) =
 103             consumedMessages.map(VesEvent::parseFrom)
 104
 105     companion object {
 106         private val logger = Logger(MessageStreamValidation::class)
 107     }
 108 }