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.domain.PayloadWireFrameMessage
  24 import org.onap.dcae.collectors.veshv.ves.message.generator.config.MessageParameters
  25 import org.onap.dcae.collectors.veshv.utils.logging.Logger
  26 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageGenerator
  27 import ratpack.exec.Promise
  28 import ratpack.handling.Chain
  29 import ratpack.handling.Context
  30 import ratpack.server.RatpackServer
  31 import ratpack.server.ServerConfig
  32 import reactor.core.publisher.Flux
  33 import reactor.core.scheduler.Schedulers
  34 import javax.json.Json
  35 import javax.json.JsonObject
  36
  37 /**
  38  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
  39  * @since June 2018
  40  */
  41 internal class HttpServer(private val vesClient: XnfSimulator) {
  42
  43     fun start(port: Int = DEFAULT_PORT): IO<RatpackServer> = IO {
  44         RatpackServer.start { server ->
  45             server.serverConfig(ServerConfig.embedded().port(port))
  46                     .handlers(this::configureHandlers)
  47         }
  48     }
  49
  50
  51     private fun configureHandlers(chain: Chain) {
  52         chain
  53                 .post("simulator/sync") { ctx ->
  54                     createMessageFlux(ctx)
  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                     createMessageFlux(ctx)
  62                             .map { vesClient.sendRx(it) }
  63                             .map { it.subscribeOn(Schedulers.elastic()).subscribe() }
  64                             .onError { handleException(it, ctx) }
  65                             .then { sendAcceptedResponse(ctx) }
  66                 }
  67     }
  68
  69     private fun createMessageFlux(ctx: Context): Promise<Flux<PayloadWireFrameMessage>> {
  70         return ctx.request.body
  71                 .map { Json.createReader(it.inputStream).readObject() }
  72                 .map { extractMessageParameters(it) }
  73                 .map { MessageGenerator.INSTANCE.createMessageFlux(it) }
  74     }
  75
  76     private fun sendAcceptedResponse(ctx: Context) {
  77         ctx.response
  78                 .status(STATUS_OK)
  79                 .send(CONTENT_TYPE_APPLICATION_JSON, Json.createObjectBuilder()
  80                         .add("response", "Request accepted")
  81                         .build()
  82                         .toString())
  83     }
  84
  85     private fun handleException(t: Throwable, ctx: Context) {
  86         logger.warn("Failed to process the request - ${t.localizedMessage}")
  87         logger.debug("Exception thrown when processing the request", t)
  88         ctx.response
  89                 .status(STATUS_BAD_REQUEST)
  90                 .send(CONTENT_TYPE_APPLICATION_JSON, Json.createObjectBuilder()
  91                         .add("response", "Request was not accepted")
  92                         .add("exception", t.localizedMessage)
  93                         .build()
  94                         .toString())
  95     }
  96
  97     private fun extractMessageParameters(request: JsonObject): MessageParameters =
  98             try {
  99                 val commonEventHeader = MessageGenerator.INSTANCE
 100                         .parseCommonHeader(request.getJsonObject("commonEventHeader"))
 101                 val messagesAmount = request.getJsonNumber("messagesAmount").longValue()
 102                 MessageParameters(commonEventHeader, messagesAmount)
 103             } catch (e: Exception) {
 104                 throw ValidationException("Validating request body failed", e)
 105             }
 106
 107
 108     companion object {
 109         private val logger = Logger(HttpServer::class)
 110         const val DEFAULT_PORT = 5000
 111         const val STATUS_OK = 200
 112         const val STATUS_BAD_REQUEST = 400
 113         const val CONTENT_TYPE_APPLICATION_JSON = "application/json"
 114     }
 115 }
 116
 117 internal class ValidationException(message: String?, cause: Exception) : Exception(message, cause)