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