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.adapters
  21
  22 import arrow.effects.IO
  23 import org.onap.dcae.collectors.veshv.simulators.dcaeapp.impl.DcaeAppSimulator
  24 import org.onap.dcae.collectors.veshv.utils.NettyServerHandle
  25 import org.onap.dcae.collectors.veshv.utils.ServerHandle
  26 import org.onap.dcae.collectors.veshv.utils.http.*
  27 import org.onap.dcae.collectors.veshv.utils.logging.Logger
  28 import reactor.core.publisher.Mono
  29 import reactor.netty.http.server.HttpServer
  30 import reactor.netty.http.server.HttpServerRoutes
  31 import java.net.InetSocketAddress
  32
  33 /**
  34  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  35  * @since May 2018
  36  */
  37 class DcaeAppApiServer(private val simulator: DcaeAppSimulator) {
  38     private val responseValid by lazy {
  39         Responses.statusResponse(
  40                 name = "valid",
  41                 message = VALID_RESPONSE_MESSAGE
  42         )
  43     }
  44
  45     private val responseInvalid by lazy {
  46         Responses.statusResponse(
  47                 name = "invalid",
  48                 message = INVALID_RESPONSE_MESSAGE,
  49                 httpStatus = HttpStatus.BAD_REQUEST
  50         )
  51     }
  52
  53
  54     fun start(socketAddress: InetSocketAddress, kafkaTopics: Set<String>): IO<ServerHandle> =
  55             simulator.listenToTopics(kafkaTopics).map {
  56                 HttpServer.create()
  57                         .host(socketAddress.hostName)
  58                         .port(socketAddress.port)
  59                         .route(::setRoutes)
  60                         .let { NettyServerHandle(it.bindNow()) }
  61             }
  62
  63
  64     private fun setRoutes(route: HttpServerRoutes) {
  65         route
  66                 .put("/configuration/topics") { req, res ->
  67                     req
  68                             .receive().aggregate().asString()
  69                             .flatMap {
  70                                 val option = simulator.listenToTopics(it)
  71                                 res.sendOrError(option).then()
  72                             }
  73                 }
  74                 .delete("/messages") { _, res ->
  75                     logger.info { "Resetting simulator state" }
  76                     res
  77                             .header("Content-type", CONTENT_TEXT)
  78                             .sendOrError(simulator.resetState())
  79                 }
  80                 .get("/messages/all/count") { _, res ->
  81                     logger.info { "Processing request for count of received messages" }
  82                     simulator.state().fold(
  83                             {
  84                                 logger.warn { "Error - number of messages could not be specified" }
  85                                 res.status(HttpConstants.STATUS_NOT_FOUND)
  86                             },
  87                             {
  88                                 logger.info { "Returned number of received messages: ${it.messagesCount}" }
  89                                 res.sendString(Mono.just(it.messagesCount.toString()))
  90                             }
  91                     )
  92                 }
  93                 .post("/messages/all/validate") { req, res ->
  94                     req
  95                             .receive().aggregate().asInputStream()
  96                             .flatMap { body ->
  97                                 logger.info { "Processing request for message validation" }
  98                                 val response =
  99                                         simulator.validate(body)
 100                                                 .map(::resolveValidationResponse)
 101                                 res.sendAndHandleErrors(response).then()
 102                             }
 103                 }
 104                 .get("/healthcheck") { _, res ->
 105                     val status = HttpConstants.STATUS_OK
 106                     logger.info { "Healthcheck OK, returning status: $status" }
 107                     res.status(status).send()
 108                 }
 109     }
 110
 111     private fun resolveValidationResponse(isValid: Boolean): Response =
 112             if (isValid) {
 113                 logger.info { "Comparison result: $VALID_RESPONSE_MESSAGE" }
 114                 responseValid
 115             } else {
 116                 logger.info { "Comparison result: $INVALID_RESPONSE_MESSAGE" }
 117                 responseInvalid
 118             }
 119
 120
 121     companion object {
 122         private const val CONTENT_TEXT = "text/plain"
 123         private const val VALID_RESPONSE_MESSAGE = "validation passed"
 124         private const val INVALID_RESPONSE_MESSAGE = "consumed messages don't match data from validation request"
 125         private val logger = Logger(DcaeAppApiServer::class)
 126     }
 127 }
 128