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