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 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)
54 * TOD0: if the code would look like:
55 * ```IO { active.set(false) }```
56 * the tests wouldn't pass even though `.unsafeRunSync()` is called (see HvVesSpec)
58 override fun close() = active.set(false).run { IO.unit }
62 * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
65 class CountingSink : Sink {
66 private val atomicCount = AtomicLong(0)
69 get() = atomicCount.get()
71 override fun send(messages: Flux<RoutedMessage>): Flux<ConsumedMessage> {
72 return messages.doOnNext {
73 atomicCount.incrementAndGet()
74 }.map(::SuccessfullyConsumedMessage)
79 open class ProcessingSink(private val transformer: (Flux<RoutedMessage>) -> Publisher<ConsumedMessage>) : Sink {
80 override fun send(messages: Flux<RoutedMessage>): Flux<ConsumedMessage> =
81 messages.transform(transformer)
84 class AlwaysSuccessfulSink : ProcessingSink({ it.map(::SuccessfullyConsumedMessage) })
86 class AlwaysFailingSink : ProcessingSink({ stream ->
87 stream.map { FailedToConsumeMessage(it, null, MessageDropCause.KAFKA_FAILURE) }
90 class DelayingSink(delay: Duration) : ProcessingSink({ it.delayElements(delay).map(::SuccessfullyConsumedMessage) })