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

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * dcaegen2-collectors-veshv
   4  * ================================================================================
   5  * Copyright (C) 2018 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.xnf.impl
  21
  22 import arrow.effects.IO
  23 import org.onap.dcae.collectors.veshv.utils.logging.Logger
  24 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageGenerator
  25 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParametersParser
  26 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParametersParser.Companion.INSTANCE
  27 import ratpack.handling.Chain
  28 import ratpack.handling.Context
  29 import ratpack.server.RatpackServer
  30 import ratpack.server.ServerConfig
  31 import reactor.core.scheduler.Schedulers
  32 import javax.json.Json
  33
  34 /**
  35  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
  36  * @since June 2018
  37  */
  38 internal class HttpServer(private val vesClient: XnfSimulator,
  39                           private val messageParametersParser: MessageParametersParser = INSTANCE) {
  40
  41     fun start(port: Int): IO<RatpackServer> = IO {
  42         RatpackServer.start { server ->
  43             server.serverConfig(ServerConfig.embedded().port(port))
  44                     .handlers(this::configureHandlers)
  45         }
  46     }
  47
  48     private fun configureHandlers(chain: Chain) {
  49         chain
  50                 .post("simulator/sync") { ctx ->
  51                     ctx.request.body
  52                             .map { Json.createReader(it.inputStream).readArray() }
  53                             .map { messageParametersParser.parse(it) }
  54                             .map { MessageGenerator.INSTANCE.createMessageFlux(it) }
  55                             .map { vesClient.sendIo(it) }
  56                             .map { it.unsafeRunSync() }
  57                             .onError { handleException(it, ctx) }
  58                             .then { sendAcceptedResponse(ctx) }
  59                 }
  60                 .post("simulator/async") { ctx ->
  61                     ctx.request.body
  62                             .map { Json.createReader(it.inputStream).readArray() }
  63                             .map { messageParametersParser.parse(it) }
  64                             .map { MessageGenerator.INSTANCE.createMessageFlux(it) }
  65                             .map { vesClient.sendRx(it) }
  66                             .map { it.subscribeOn(Schedulers.elastic()).subscribe() }
  67                             .onError { handleException(it, ctx) }
  68                             .then { sendAcceptedResponse(ctx) }
  69                 }
  70                 .get("healthcheck") { ctx ->
  71                     ctx.response.status(STATUS_OK).send()
  72                 }
  73     }
  74
  75     private fun sendAcceptedResponse(ctx: Context) {
  76         ctx.response
  77                 .status(STATUS_OK)
  78                 .send(CONTENT_TYPE_APPLICATION_JSON, Json.createObjectBuilder()
  79                         .add("response", "Request accepted")
  80                         .build()
  81                         .toString())
  82     }
  83
  84     private fun handleException(t: Throwable, ctx: Context) {
  85         logger.warn("Failed to process the request - ${t.localizedMessage}")
  86         logger.debug("Exception thrown when processing the request", t)
  87         ctx.response
  88                 .status(STATUS_BAD_REQUEST)
  89                 .send(CONTENT_TYPE_APPLICATION_JSON, Json.createObjectBuilder()
  90                         .add("response", "Request was not accepted")
  91                         .add("exception", t.localizedMessage)
  92                         .build()
  93                         .toString())
  94     }
  95
  96     companion object {
  97         private val logger = Logger(HttpServer::class)
  98         const val STATUS_OK = 200
  99         const val STATUS_BAD_REQUEST = 400
 100         const val CONTENT_TYPE_APPLICATION_JSON = "application/json"
 101     }
 102 }