sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/NettyTcpServer.kt

   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.impl.socket
  21
  22 import arrow.core.getOrElse
  23 import arrow.effects.IO
  24 import org.onap.dcae.collectors.veshv.boundary.CollectorProvider
  25 import org.onap.dcae.collectors.veshv.boundary.Server
  26 import org.onap.dcae.collectors.veshv.model.ClientContext
  27 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.info
  28 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.debug
  29 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.withWarn
  30 import org.onap.dcae.collectors.veshv.utils.logging.Marker
  31 import org.onap.dcae.collectors.veshv.model.ServerConfiguration
  32 import org.onap.dcae.collectors.veshv.ssl.boundary.ServerSslContextFactory
  33 import org.onap.dcae.collectors.veshv.utils.NettyServerHandle
  34 import org.onap.dcae.collectors.veshv.utils.ServerHandle
  35 import org.onap.dcae.collectors.veshv.utils.logging.Logger
  36 import reactor.core.publisher.Mono
  37 import reactor.netty.ByteBufFlux
  38 import reactor.netty.Connection
  39 import reactor.netty.NettyInbound
  40 import reactor.netty.NettyOutbound
  41 import reactor.netty.tcp.TcpServer
  42 import java.time.Duration
  43
  44 /**
  45  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  46  * @since May 2018
  47  */
  48 internal class NettyTcpServer(private val serverConfig: ServerConfiguration,
  49                               private val sslContextFactory: ServerSslContextFactory,
  50                               private val collectorProvider: CollectorProvider) : Server {
  51
  52     override fun start(): IO<ServerHandle> = IO {
  53         val tcpServer = TcpServer.create()
  54                 .addressSupplier { serverConfig.serverListenAddress }
  55                 .configureSsl()
  56                 .handle(this::handleConnection)
  57
  58         NettyServerHandle(tcpServer.bindNow())
  59     }
  60
  61     private fun TcpServer.configureSsl() =
  62             sslContextFactory
  63                     .createSslContext(serverConfig.securityConfiguration)
  64                     .map { sslContext ->
  65                         logger.info { "Collector configured with SSL enabled" }
  66                         this.secure { b -> b.sslContext(sslContext) }
  67                     }.getOrElse {
  68                         logger.info { "Collector configured with SSL disabled" }
  69                         this
  70                     }
  71
  72     private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> {
  73         val clientContext = ClientContext(nettyOutbound.alloc())
  74         nettyInbound.withConnection {
  75             clientContext.clientAddress = it.address()
  76         }
  77
  78         logger.debug(clientContext::asMap, Marker.ENTRY) { "Client connection request received" }
  79         return collectorProvider(clientContext).fold(
  80                 {
  81                     logger.warn(clientContext::asMap) { "Collector not ready. Closing connection..." }
  82                     Mono.empty()
  83                 },
  84                 {
  85                     logger.info(clientContext::asMap) { "Handling new connection" }
  86                     nettyInbound.withConnection { conn ->
  87                         conn.configureIdleTimeout(clientContext, serverConfig.idleTimeout)
  88                                 .logConnectionClosed(clientContext)
  89                     }
  90                     it.handleConnection(createDataStream(nettyInbound))
  91                 }
  92         )
  93     }
  94
  95     private fun createDataStream(nettyInbound: NettyInbound): ByteBufFlux = nettyInbound
  96             .receive()
  97             .retain()
  98
  99     private fun Connection.configureIdleTimeout(ctx: ClientContext, timeout: Duration): Connection {
 100         onReadIdle(timeout.toMillis()) {
 101             logger.info(ctx) {
 102                 "Idle timeout of ${timeout.seconds} s reached. Closing connection from ${address()}..."
 103             }
 104             disconnectClient(ctx)
 105         }
 106         return this
 107     }
 108
 109     private fun Connection.disconnectClient(ctx: ClientContext) {
 110         channel().close().addListener {
 111             logger.debug(ctx::asMap, Marker.EXIT) { "Closing client channel." }
 112             if (it.isSuccess)
 113                 logger.debug(ctx) { "Channel closed successfully." }
 114             else
 115                 logger.withWarn(ctx) { log("Channel close failed", it.cause()) }
 116         }
 117     }
 118
 119     private fun Connection.logConnectionClosed(ctx: ClientContext): Connection {
 120         onTerminate().subscribe {
 121             // TODO: this code is never executed (at least with ssl-enabled, did not checked with ssl-disabled)
 122             logger.info(ctx::asMap, Marker.EXIT) { "Connection has been closed" }
 123         }
 124         return this
 125     }
 126
 127     companion object {
 128         private val logger = Logger(NettyTcpServer::class)
 129     }
 130 }