diff --git a/src/main/kotlin/Main.kt b/src/main/kotlin/Main.kt index 18c0de6..497b3aa 100644 --- a/src/main/kotlin/Main.kt +++ b/src/main/kotlin/Main.kt @@ -1,68 +1,97 @@ import io.reactivex.rxjava3.core.Completable import io.reactivex.rxjava3.core.Flowable import io.reactivex.rxjava3.core.Maybe +import io.reactivex.rxjava3.core.Single import java.rmi.server.ServerNotActiveException -import java.util.* +import java.time.LocalTime import java.util.concurrent.TimeUnit import kotlin.random.Random private const val LATENCY = 700L private const val RESPONSE_LENGTH = 2048 +// Функции можно вызывать отсюда для проверки +// для ДЗ лучше использовать blockingSubscribe вместо subscribe потому что subscribe подпишется на изменения, +// но изменения в большинстве случаев будут получены позже, чем выполнится функция main, поэтому в консоли ничего +// не будет выведено. blockingSubscribe работает синхронно, поэтому результат будет выведен в консоль +// +// В реальных программах нужно использовать subscribe или передавать данные от источника к источнику для +// асинхронного выполнения кода. +// +// Несмотря на то, что в некоторых заданиях фигурируют слова "синхронный" и "асинхронный" в рамках текущего ДЗ +// это всего лишь имитация, реальное переключение между потоками будет рассмотрено на следующем семинаре + fun main() { - // Функции можно вызывать отсюда для проверки - // для ДЗ лучше использовать blockingSubscribe вместо subscribe потому что subscribe подпишется на изменения, - // но изменения в большинстве случаев будут получены позже, чем выполнится функция main, поэтому в консоли ничего - // не будет выведено. blockingSubscribe работает синхронно, поэтому результат будет выведен в консоль - // - // В реальных программах нужно использовать subscribe или передавать данные от источника к источнику для - // асинхронного выполнения кода. - // - // Несмотря на то, что в некоторых заданиях фигурируют слова "синхронный" и "асинхронный" в рамках текущего ДЗ - // это всего лишь имитация, реальное переключение между потоками будет рассмотрено на следующем семинаре + println("1. requestDataFromServerAsync") + requestDataFromServerAsync(true).blockingSubscribe({ println("size: ${it.size}") }, { println("error") }) + requestDataFromServerAsync(false).blockingSubscribe({ println("size: ${it.size}") }, { println("error") }) + + println("2. requestServerAsync") + requestServerAsync(true).blockingSubscribe({ println("complete") }, { println("error") }) + requestServerAsync(false).blockingSubscribe({ println("complete") }, { println("error") }) + + println("3. requestDataFromDbAsync") + requestDataFromDbAsync(true).blockingSubscribe({ println("$it") }, { println("error") }) + requestDataFromDbAsync(false).blockingSubscribe({ println("$it") }, { println("error") }, { println("complete") }) + + println("4. emitEachSecond") + emitEachSecond() + + println("5. xMap") + xMap { flatMapCompletable(it) } + xMap { concatMapCompletable(it) } + xMap { switchMapCompletable(it) } } // 1) Какой источник лучше всего подойдёт для запроса на сервер, который возвращает результат? // Почему? // Дописать функцию -fun requestDataFromServerAsync() /* -> ??? */ { +fun requestDataFromServerAsync(success: Boolean): Single /* -> ??? */ { // Функция имитирует синхронный запрос на сервер, возвращающий результат fun getDataFromServerSync(): ByteArray? { - Thread.sleep(LATENCY); - val success = Random.nextBoolean() + Thread.sleep(LATENCY) + // val success = Random.nextBoolean() return if (success) Random.nextBytes(RESPONSE_LENGTH) else null } /* return ??? */ + return Single.fromCallable { getDataFromServerSync() } } // 2) Какой источник лучше всего подойдёт для запроса на сервер, который НЕ возвращает результат? // Почему? // Дописать функцию -fun requestServerAsync() /* -> ??? */ { +fun requestServerAsync(success: Boolean): Completable /* -> ??? */ { // Функция имитирует синхронный запрос на сервер, не возвращающий результат fun getDataFromServerSync() { Thread.sleep(LATENCY) - if (Random.nextBoolean()) throw ServerNotActiveException() + if (!success) throw ServerNotActiveException() } /* return ??? */ + return Completable.fromCallable { getDataFromServerSync() } } // 3) Какой источник лучше всего подойдёт для однократного асинхронного возвращения значения из базы данных? // Почему? // Дописать функцию -fun requestDataFromDbAsync() /* -> ??? */ { +fun requestDataFromDbAsync(success: Boolean): Maybe /* -> ??? */ { // Функция имитирует синхронный запрос к БД не возвращающий результата - fun getDataFromDbSync(): T? { - Thread.sleep(LATENCY); return null + fun getDataFromDbSync(): String? { + Thread.sleep(LATENCY) + return if (success) "DATA" else null } /* return */ + return Maybe.fromCallable { getDataFromDbSync() } +} + +fun now(): LocalTime { + return LocalTime.now() } // 4) Примените к источнику оператор (несколько операторов), которые приведут к тому, чтобы элемент из источника @@ -75,9 +104,21 @@ fun emitEachSecond() { fun source(): Flowable = Flowable.interval(500, TimeUnit.MILLISECONDS) // Принтер - fun printer(value: Long) = println("${Date()}: value = $value") + fun printer(value: Long) = println("${now()}: value = $value") + println("${now()}: start") // code here + source() + .onBackpressureBuffer() // Чтоб не упало + .takeWhile { it <= 10 } // Чтоб завершилось + + .concatMapSingle { v -> Single.timer(1, TimeUnit.SECONDS).map { v } } + // или по тупому + // .doAfterNext { Thread.sleep(1 * 1000) } + + .doOnError { println(it.localizedMessage) } + .doOnComplete { println("complete") } + .blockingSubscribe { printer(it) } } // 5) Функция для изучения разницы между операторами concatMap, flatMap, switchMap @@ -99,12 +140,13 @@ fun xMap(mapper: Flowable.(internalMapper: (Int) -> Completable) -> Complet fun waitOneSecond() = Completable.timer(1, TimeUnit.SECONDS) - println("${Date()}: start") - Flowable.fromIterable(0..20) + println("${now()}: start") + Flowable.fromIterable(0..10) + .doOnNext { println("${now()}: index: $it") } .mapper { iterableIndex -> waitOneSecond() - .doOnComplete { println("${Date()}: finished operation for iterable index $iterableIndex") } + .doOnComplete { println("${now()}: finished operation for iterable index $iterableIndex") } } .blockingSubscribe()