2 * ============LICENSE_START=======================================================
3 * dcaegen2-collectors-veshv
4 * ================================================================================
5 * Copyright (C) 2018-2019 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
11 * http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
20 package org.onap.dcae.collectors.veshv.tests.fakes
22 import org.onap.dcae.collectors.veshv.boundary.Sink
23 import org.onap.dcae.collectors.veshv.model.ConsumedMessage
24 import org.onap.dcae.collectors.veshv.model.FailedToConsumeMessage
25 import org.onap.dcae.collectors.veshv.model.MessageDropCause
26 import org.onap.dcae.collectors.veshv.domain.RoutedMessage
27 import org.onap.dcae.collectors.veshv.model.SuccessfullyConsumedMessage
28 import org.reactivestreams.Publisher
29 import reactor.core.publisher.Flux
30 import reactor.core.publisher.Mono
31 import java.time.Duration
33 import java.util.concurrent.ConcurrentLinkedDeque
34 import java.util.concurrent.atomic.AtomicBoolean
35 import java.util.concurrent.atomic.AtomicLong
38 * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
41 class StoringSink : Sink {
42 private val sent: Deque<RoutedMessage> = ConcurrentLinkedDeque()
43 private val active = AtomicBoolean(true)
44 val closed get() = !active.get()
46 val sentMessages: List<RoutedMessage>
49 override fun send(messages: Flux<RoutedMessage>): Flux<ConsumedMessage> {
50 return messages.doOnNext(sent::addLast).map(::SuccessfullyConsumedMessage)
53 override fun close(): Mono<Void> = Mono.fromRunnable {
59 * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
62 class CountingSink : Sink {
63 private val atomicCount = AtomicLong(0)
66 get() = atomicCount.get()
68 override fun send(messages: Flux<RoutedMessage>): Flux<ConsumedMessage> {
69 return messages.doOnNext {
70 atomicCount.incrementAndGet()
71 }.map(::SuccessfullyConsumedMessage)
76 open class ProcessingSink(private val transformer: (Flux<RoutedMessage>) -> Publisher<ConsumedMessage>) : Sink {
77 override fun send(messages: Flux<RoutedMessage>): Flux<ConsumedMessage> =
78 messages.transform(transformer)
81 class AlwaysSuccessfulSink : ProcessingSink({ it.map(::SuccessfullyConsumedMessage) })
83 class AlwaysFailingSink : ProcessingSink({ stream ->
84 stream.map { FailedToConsumeMessage(it, null, MessageDropCause.KAFKA_FAILURE) }
87 class DelayingSink(delay: Duration) : ProcessingSink({ it.delayElements(delay).map(::SuccessfullyConsumedMessage) })