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.dcaeapp.kafka
  21
  22 import reactor.core.publisher.Mono
  23 import reactor.kafka.receiver.ReceiverRecord
  24
  25 /**
  26  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  27  * @since June 2018
  28  */
  29
  30 class ConsumerState(val msgCount: Long, val lastKey: ByteArray?, val lastValue: ByteArray?)
  31
  32 interface ConsumerStateProvider {
  33     fun currentState(): Mono<ConsumerState>
  34 }
  35
  36 class Consumer : ConsumerStateProvider {
  37     private var msgCount = 0L
  38     private var lastKey: ByteArray? = null
  39     private var lastValue: ByteArray? = null
  40
  41     override fun currentState(): Mono<ConsumerState> = Mono.create { sink ->
  42         val state = synchronized(this) {
  43             ConsumerState(msgCount, lastKey, lastValue)
  44         }
  45         sink.success(state)
  46     }
  47
  48     fun update(record: ReceiverRecord<ByteArray, ByteArray>) {
  49         synchronized(this) {
  50             msgCount++
  51             lastKey = record.key()
  52             lastValue = record.value()
  53         }
  54     }
  55 }