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 arrow.effects.IO
23 import org.onap.dcae.collectors.veshv.boundary.Sink
24 import org.onap.dcae.collectors.veshv.model.ConsumedMessage
25 import org.onap.dcae.collectors.veshv.model.FailedToConsumeMessage
26 import org.onap.dcae.collectors.veshv.model.MessageDropCause
27 import org.onap.dcae.collectors.veshv.domain.RoutedMessage
28 import org.onap.dcae.collectors.veshv.model.SuccessfullyConsumedMessage
29 import org.reactivestreams.Publisher
30 import reactor.core.publisher.Flux
31 import reactor.core.publisher.Mono
32 import java.time.Duration
34 import java.util.concurrent.ConcurrentLinkedDeque
35 import java.util.concurrent.atomic.AtomicBoolean
36 import java.util.concurrent.atomic.AtomicLong
39 * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
42 class StoringSink : Sink {
43 private val sent: Deque<RoutedMessage> = ConcurrentLinkedDeque()
44 private val active = AtomicBoolean(true)
45 val closed get() = !active.get()
47 val sentMessages: List<RoutedMessage>
50 override fun send(messages: Flux<RoutedMessage>): Flux<ConsumedMessage> {
51 return messages.doOnNext(sent::addLast).map(::SuccessfullyConsumedMessage)
54 override fun close(): Mono<Void> = Mono.fromRunnable {
60 * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
63 class CountingSink : Sink {
64 private val atomicCount = AtomicLong(0)
67 get() = atomicCount.get()
69 override fun send(messages: Flux<RoutedMessage>): Flux<ConsumedMessage> {
70 return messages.doOnNext {
71 atomicCount.incrementAndGet()
72 }.map(::SuccessfullyConsumedMessage)
77 open class ProcessingSink(private val transformer: (Flux<RoutedMessage>) -> Publisher<ConsumedMessage>) : Sink {
78 override fun send(messages: Flux<RoutedMessage>): Flux<ConsumedMessage> =
79 messages.transform(transformer)
82 class AlwaysSuccessfulSink : ProcessingSink({ it.map(::SuccessfullyConsumedMessage) })
84 class AlwaysFailingSink : ProcessingSink({ stream ->
85 stream.map { FailedToConsumeMessage(it, null, MessageDropCause.KAFKA_FAILURE) }
88 class DelayingSink(delay: Duration) : ProcessingSink({ it.delayElements(delay).map(::SuccessfullyConsumedMessage) })