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

   1 package org.onap.dcae.collectors.veshv.main.impl
   2
   3 import io.netty.buffer.ByteBufAllocator
   4 import org.onap.dcae.collectors.veshv.domain.WireFrame
   5 import org.onap.dcae.collectors.veshv.main.config.ClientConfiguration
   6 import org.onap.dcae.collectors.veshv.utils.logging.Logger
   7 import org.reactivestreams.Publisher
   8 import reactor.core.publisher.Flux
   9 import reactor.ipc.netty.NettyInbound
  10 import reactor.ipc.netty.NettyOutbound
  11 import reactor.ipc.netty.tcp.TcpClient
  12 import java.util.function.BiFunction
  13
  14
  15 /**
  16  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
  17  * @since June 2018
  18  */
  19 class VesHvClient(configuration: ClientConfiguration) {
  20
  21     private val logger = Logger(VesHvClient::class)
  22     private val client: TcpClient = TcpClient.create(configuration.vesHost, configuration.vesPort)
  23
  24     fun send(messages: Flux<WireFrame>) {
  25         client.start(BiFunction { i, o -> handler(i, o, messages) })
  26     }
  27
  28     // sending flux with multiple WireFrames not supported yet
  29     private fun handler(nettyInbound: NettyInbound,
  30                         nettyOutbound: NettyOutbound,
  31                         messages: Flux<WireFrame>): Publisher<Void> {
  32
  33         nettyInbound
  34                 .receive()
  35                 .asString(Charsets.UTF_8)
  36                 .subscribe { str -> logger.info("Server response: $str") }
  37
  38         return nettyOutbound
  39                 .options { it.flushOnEach() }
  40                 .send(messages.map { it.encode(ByteBufAllocator.DEFAULT) })
  41     }
  42 }