App.kt

package com.github.eutkin

import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.io.InputStream
import java.math.BigDecimal
import java.math.MathContext
import java.time.Duration
import java.util.concurrent.ThreadLocalRandom
import java.util.concurrent.atomic.AtomicBoolean

object App {

    @JvmStatic
    fun main(args: Array<String>) {
        run(System.`in`)
    }

    /**
     * На вход поступают денежные пары (USD/RUB), разделенные между собой пробельными символами.
     *
     * Каждую пару мы отправляем в поисковый сервис
     * и как только получаем ответ по любой из пар, прекращаем работу.
     */
    private fun run(input: InputStream) {
        val stop = AtomicBoolean(false)
        Flux
            .create<String> { sink ->
                input.bufferedReader().use { reader ->
                    while (!stop.get()) {
                        val line = reader.readLine()
                        if (stop.get()) {
                            sink.complete()
                            break
                        } else {
                            sink.next(line)
                        }
                    }
                }
            }
            .map { line -> line.split("\\s+".toRegex()) }
            .map { (key, value) -> key to value }
            .flatMap { (key, value) -> sendToServiceA(key, value) }
            .doOnNext {
                if (it == null) {
                    throw IllegalStateException("Rate must be not null")
                } else {
                    println(it)
                }
            }
            .then(
               Mono.just(stop.set(true))
            )
            .block()
    }

    private fun sendToServiceA(key: String, value: String): Mono<BigDecimal> =
        Mono
            .fromSupplier{ ThreadLocalRandom.current().nextDouble(0.0, 10.0).toBigDecimal(MathContext(2))}
            .delayElement(Duration.ofMillis(ThreadLocalRandom.current().nextLong(500, 2000)))


}