sources/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/adapters/DcaeAppApiServer.kt

   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     fun start(socketAddress: InetSocketAddress, kafkaTopics: Set<String>): IO<ServerHandle> =
  54             IO {
  55                 simulator.listenToTopics(kafkaTopics)
  56                 HttpServer.create()
  57                         .host(socketAddress.hostName)
  58                         .port(socketAddress.port)
  59                         .route(::setRoutes)
  60                         .let { NettyServerHandle(it.bindNow()) }
  61             }
  62
  63     private fun setRoutes(route: HttpServerRoutes) {
  64         route
  65                 .put("/configuration/topics") { req, res ->
  66                     req
  67                             .receive().aggregate().asString()
  68                             .flatMap {
  69                                res.sendOrError{ simulator.listenToTopics(it) }
  70                             }
  71                 }
  72                 .delete("/messages") { _, res ->
  73                     logger.info { "Resetting simulator state" }
  74
  75                     res
  76                             .header("Content-type", CONTENT_TEXT)
  77                             .sendOrError { simulator.resetState() }
  78                 }
  79                 .get("/messages/all/count") { _, res ->
  80                     logger.info { "Processing request for count of received messages" }
  81                     simulator.state().fold(
  82                             {
  83                                 logger.warn { "Error - number of messages could not be specified" }
  84                                 res.status(HttpConstants.STATUS_NOT_FOUND)
  85                             },
  86                             {
  87                                 logger.info { "Returned number of received messages: ${it.messagesCount}" }
  88                                 res.sendString(Mono.just(it.messagesCount.toString()))
  89                             }
  90                     )
  91                 }
  92                 .post("/messages/all/validate") { req, res ->
  93                     req
  94                             .receive().aggregate().asInputStream()
  95                             .map {
  96                                 logger.info { "Processing request for message validation" }
  97                                 simulator.validate(it)
  98                                         .map(::resolveValidationResponse)
  99                             }
 100                             .flatMap {
 101                                 res.sendAndHandleErrors(it)
 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