Remove outdated reactive guide and add section about Reactive Streams to flow.md
diff --git a/README.md b/README.md
index e31d6f2..862ce16 100644
--- a/README.md
+++ b/README.md
@@ -62,7 +62,6 @@
* Guides and manuals:
* [Guide to kotlinx.coroutines by example](https://kotlinlang.org/docs/reference/coroutines/coroutines-guide.html) (**read it first**)
* [Guide to UI programming with coroutines](ui/coroutines-guide-ui.md)
- * [Guide to reactive streams with coroutines](reactive/coroutines-guide-reactive.md)
* [Debugging capabilities in kotlinx.coroutines](docs/debugging.md)
* [Compatibility policy and experimental annotations](docs/compatibility.md)
* [Change log for kotlinx.coroutines](CHANGES.md)
diff --git a/coroutines-guide.md b/coroutines-guide.md
index 4a7fb67..bebd0d0 100644
--- a/coroutines-guide.md
+++ b/coroutines-guide.md
@@ -79,6 +79,7 @@
* <a name='upstream-exceptions-only'></a>[Upstream exceptions only](docs/flow.md#upstream-exceptions-only)
* <a name='imperative-versus-declarative'></a>[Imperative versus declarative](docs/flow.md#imperative-versus-declarative)
* <a name='launching-flow'></a>[Launching flow](docs/flow.md#launching-flow)
+ * <a name='flow-and-reactive-streams'></a>[Flow and Reactive Streams](docs/flow.md#flow-and-reactive-streams)
<!--- TOC_REF docs/channels.md -->
* <a name='channels'></a>[Channels](docs/channels.md#channels)
* <a name='channel-basics'></a>[Channel basics](docs/channels.md#channel-basics)
diff --git a/docs/coroutines-guide.md b/docs/coroutines-guide.md
index 5f41d06..e3f18d2 100644
--- a/docs/coroutines-guide.md
+++ b/docs/coroutines-guide.md
@@ -28,6 +28,5 @@
## Additional references
* [Guide to UI programming with coroutines](../ui/coroutines-guide-ui.md)
-* [Guide to reactive streams with coroutines](../reactive/coroutines-guide-reactive.md)
* [Coroutines design document (KEEP)](https://github.com/Kotlin/kotlin-coroutines/blob/master/kotlin-coroutines-informal.md)
* [Full kotlinx.coroutines API reference](https://kotlin.github.io/kotlinx.coroutines)
diff --git a/docs/flow.md b/docs/flow.md
index 2a3ffb5..7837a54 100644
--- a/docs/flow.md
+++ b/docs/flow.md
@@ -58,6 +58,7 @@
* [Upstream exceptions only](#upstream-exceptions-only)
* [Imperative versus declarative](#imperative-versus-declarative)
* [Launching flow](#launching-flow)
+ * [Flow and Reactive Streams](#flow-and-reactive-streams)
<!--- END_TOC -->
@@ -1794,6 +1795,18 @@
Note that [launchIn] also returns a [Job], which can be used to [cancel][Job.cancel] the corresponding flow collection
coroutine only without cancelling the whole scope or to [join][Job.join] it.
+### Flow and Reactive Streams
+
+For those who are familiar with [Reactive Streams](https://www.reactive-streams.org/) or reactive frameworks such as RxJava and project Reactor,
+design of the Flow may look very familiar.
+
+Indeed, its design was inspired by Reactive Streams and its various implementations. But Flow main goal is to have as simple design as possible,
+be Kotlin and suspension friendly and respect structured concurrency. Achieving this goal would be impossible without reactive pioneers and their tremendous work. You can read the complete story in [Reactive Streams and Kotlin Flows](https://medium.com/@elizarov/reactive-streams-and-kotlin-flows-bfd12772cda4) article.
+
+While being different, conceptually, Flow *is* a reactive stream and it is possible to convert it to the reactive (spec and TCK compliant) Publisher and vice versa.
+Such converters are provided by `kotlinx.coroutines` out-of-the-box and can be found in corresponding reactive modules (`kotlinx-coroutines-reactive` for Reactive Streams, `kotlinx-coroutines-reactor` for Project Reactor and `kotlinx-coroutines-rx2` for RxJava2).
+Integration modules include conversions from and to `Flow`, integration with Reactor's `Context` and suspension-friendly ways to work with various reactive entities.
+
<!-- stdlib references -->
[collections]: https://kotlinlang.org/docs/reference/collections-overview.html
diff --git a/reactive/coroutines-guide-reactive.md b/reactive/coroutines-guide-reactive.md
deleted file mode 100644
index 0eff27b..0000000
--- a/reactive/coroutines-guide-reactive.md
+++ /dev/null
@@ -1,1078 +0,0 @@
-<!--- INCLUDE .*/example-reactive-([a-z]+)-([0-9]+)\.kt
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.$$1$$2
-
--->
-<!--- KNIT kotlinx-coroutines-rx2/test/guide/.*\.kt -->
-<!--- TEST_OUT kotlinx-coroutines-rx2/test/guide/test/GuideReactiveTest.kt
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.test
-
-import kotlinx.coroutines.guide.test.*
-import org.junit.Test
-
-class GuideReactiveTest : ReactiveTestBase() {
--->
-
-# Guide to reactive streams with coroutines
-
-This guide explains the key differences between Kotlin coroutines and reactive streams and shows
-how they can be used together for the greater good. Prior familiarity with the basic coroutine concepts
-that are covered in [Guide to kotlinx.coroutines](../docs/coroutines-guide.md) is not required,
-but is a big plus. If you are familiar with reactive streams, you may find this guide
-a better introduction into the world of coroutines.
-
-There are several modules in `kotlinx.coroutines` project that are related to reactive streams:
-
-* [kotlinx-coroutines-reactive](kotlinx-coroutines-reactive) -- utilities for [Reactive Streams](https://www.reactive-streams.org)
-* [kotlinx-coroutines-reactor](kotlinx-coroutines-reactor) -- utilities for [Reactor](https://projectreactor.io)
-* [kotlinx-coroutines-rx2](kotlinx-coroutines-rx2) -- utilities for [RxJava 2.x](https://github.com/ReactiveX/RxJava)
-
-This guide is mostly based on [Reactive Streams](https://www.reactive-streams.org) specification and uses
-its `Publisher` interface with some examples based on [RxJava 2.x](https://github.com/ReactiveX/RxJava),
-which implements reactive streams specification.
-
-You are welcome to clone
-[`kotlinx.coroutines` project](https://github.com/Kotlin/kotlinx.coroutines)
-from GitHub to your workstation in order to
-run all the presented examples. They are contained in
-[reactive/kotlinx-coroutines-rx2/test/guide](kotlinx-coroutines-rx2/test/guide)
-directory of the project.
-
-## Table of contents
-
-<!--- TOC -->
-
-* [Differences between reactive streams and channels](#differences-between-reactive-streams-and-channels)
- * [Basics of iteration](#basics-of-iteration)
- * [Subscription and cancellation](#subscription-and-cancellation)
- * [Backpressure](#backpressure)
- * [Rx Subject vs BroadcastChannel](#rx-subject-vs-broadcastchannel)
-* [Operators](#operators)
- * [Range](#range)
- * [Fused filter-map hybrid](#fused-filter-map-hybrid)
- * [Take until](#take-until)
- * [Merge](#merge)
-* [Coroutine context](#coroutine-context)
- * [Threads with Rx](#threads-with-rx)
- * [Threads with coroutines](#threads-with-coroutines)
- * [Rx observeOn](#rx-observeon)
- * [Coroutine context to rule them all](#coroutine-context-to-rule-them-all)
- * [Unconfined context](#unconfined-context)
-
-<!--- END_TOC -->
-
-## Differences between reactive streams and channels
-
-This section outlines key differences between reactive streams and coroutine-based channels.
-
-### Basics of iteration
-
-The [Channel] is somewhat similar concept to the following reactive stream classes:
-
-* Reactive stream [Publisher](https://github.com/reactive-streams/reactive-streams-jvm/blob/master/api/src/main/java/org/reactivestreams/Publisher.java);
-* Rx Java 1.x [Observable](https://reactivex.io/RxJava/javadoc/rx/Observable.html);
-* Rx Java 2.x [Flowable](https://reactivex.io/RxJava/2.x/javadoc/), which implements `Publisher`.
-
-They all describe an asynchronous stream of elements (aka items in Rx), either infinite or finite,
-and all of them support backpressure.
-
-However, the `Channel` always represents a _hot_ stream of items, using Rx terminology. Elements are being sent
-into the channel by producer coroutines and are received from it by consumer coroutines.
-Every [receive][ReceiveChannel.receive] invocation consumes an element from the channel.
-Let us illustrate it with the following example:
-
-<!--- INCLUDE
-import kotlinx.coroutines.*
-import kotlinx.coroutines.channels.*
-import kotlin.coroutines.*
--->
-
-```kotlin
-fun main() = runBlocking<Unit> {
- // create a channel that produces numbers from 1 to 3 with 200ms delays between them
- val source = produce<Int> {
- println("Begin") // mark the beginning of this coroutine in output
- for (x in 1..3) {
- delay(200) // wait for 200ms
- send(x) // send number x to the channel
- }
- }
- // print elements from the source
- println("Elements:")
- source.consumeEach { // consume elements from it
- println(it)
- }
- // print elements from the source AGAIN
- println("Again:")
- source.consumeEach { // consume elements from it
- println(it)
- }
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-basic-01.kt).
-
-This code produces the following output:
-
-```text
-Elements:
-Begin
-1
-2
-3
-Again:
-```
-
-<!--- TEST -->
-
-Notice how the "Begin" line was printed just once, because the [produce] _coroutine builder_, when it is executed,
-launches one coroutine to produce a stream of elements. All the produced elements are consumed
-with [ReceiveChannel.consumeEach][consumeEach]
-extension function. There is no way to receive the elements from this
-channel again. The channel is closed when the producer coroutine is over and an attempt to receive
-from it again cannot receive anything.
-
-Let us rewrite this code using the [publish] coroutine builder from `kotlinx-coroutines-reactive` module
-instead of [produce] from `kotlinx-coroutines-core` module. The code stays the same,
-but where `source` used to have the [ReceiveChannel] type, it now has the reactive streams'
-[Publisher](https://www.reactive-streams.org/reactive-streams-1.0.0-javadoc/org/reactivestreams/Publisher.html)
-type, and where [consumeEach] was used to _consume_ elements from the channel,
-now [collect][org.reactivestreams.Publisher.collect] is used to _collect_ elements from the publisher.
-
-<!--- INCLUDE
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import kotlin.coroutines.*
--->
-
-```kotlin
-fun main() = runBlocking<Unit> {
- // create a publisher that produces numbers from 1 to 3 with 200ms delays between them
- val source = publish<Int> {
- // ^^^^^^^ <--- Difference from the previous examples is here
- println("Begin") // mark the beginning of this coroutine in output
- for (x in 1..3) {
- delay(200) // wait for 200ms
- send(x) // send number x to the channel
- }
- }
- // print elements from the source
- println("Elements:")
- source.collect { // collect elements from it
- println(it)
- }
- // print elements from the source AGAIN
- println("Again:")
- source.collect { // collect elements from it
- println(it)
- }
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-basic-02.kt).
-
-Now the output of this code changes to:
-
-```text
-Elements:
-Begin
-1
-2
-3
-Again:
-Begin
-1
-2
-3
-```
-
-<!--- TEST -->
-
-This example highlights the key difference between a reactive stream and a channel. A reactive stream is a higher-order
-functional concept. While the channel _is_ a stream of elements, the reactive stream defines a recipe on how the stream of
-elements is produced. It becomes the actual stream of elements when _collected_. Each collector may receive the same or
-a different stream of elements, depending on how the corresponding implementation of `Publisher` works.
-
-The [publish] coroutine builder used in the above example does not launch a coroutine,
-but every [collect][org.reactivestreams.Publisher.collect] invocation does.
-There are two of them here and that is why we see "Begin" printed twice.
-
-In Rx lingo, this kind of publisher is called _cold_. Many standard Rx operators produce cold streams, too. We can collect
-them from a coroutine, and every collector gets the same stream of elements.
-
-> Note that we can replicate the same behaviour that we saw with channels by using Rx
-[publish](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#publish())
-operator and [connect](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/flowables/ConnectableFlowable.html#connect())
-method with it.
-
-### Subscription and cancellation
-
-In the second example from the previous section, `source.collect { ... }` was used to collect all elements.
-Instead, we can open a channel using [openSubscription][org.reactivestreams.Publisher.openSubscription]
-and iterate over it. In this way, we can have finer-grained control over our iteration
-(using `break`, for example), as shown below:
-
-<!--- INCLUDE
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.channels.*
-import kotlinx.coroutines.reactive.*
--->
-
-```kotlin
-fun main() = runBlocking<Unit> {
- val source = Flowable.range(1, 5) // a range of five numbers
- .doOnSubscribe { println("OnSubscribe") } // provide some insight
- .doOnComplete { println("OnComplete") } // ...
- .doFinally { println("Finally") } // ... into what's going on
- var cnt = 0
- source.openSubscription().consume { // open channel to the source
- for (x in this) { // iterate over the channel to receive elements from it
- println(x)
- if (++cnt >= 3) break // break when 3 elements are printed
- }
- // Note: `consume` cancels the channel when this block of code is complete
- }
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-basic-03.kt).
-
-It produces the following output:
-
-```text
-OnSubscribe
-1
-2
-3
-Finally
-```
-
-<!--- TEST -->
-
-With an explicit `openSubscription` we should [cancel][ReceiveChannel.cancel] the corresponding
-subscription to unsubscribe from the source, but there is no need to call `cancel` explicitly --
-[consume] does that for us under the hood.
-The installed
-[doFinally](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#doFinally(io.reactivex.functions.Action))
-listener prints "Finally" to confirm that the subscription is actually being closed. Note that "OnComplete"
-is never printed because we did not consume all of the elements.
-
-We do not need to use an explicit `cancel` either if we `collect` all the elements:
-
-<!--- INCLUDE
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import kotlin.coroutines.*
--->
-
-```kotlin
-fun main() = runBlocking<Unit> {
- val source = Flowable.range(1, 5) // a range of five numbers
- .doOnSubscribe { println("OnSubscribe") } // provide some insight
- .doOnComplete { println("OnComplete") } // ...
- .doFinally { println("Finally") } // ... into what's going on
- // collect the source fully
- source.collect { println(it) }
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-basic-04.kt).
-
-We get the following output:
-
-```text
-OnSubscribe
-1
-2
-3
-OnComplete
-Finally
-4
-5
-```
-
-<!--- TEST -->
-
-Notice how "OnComplete" and "Finally" are printed before the lasts elements "4" and "5".
-It happens because our `main` function in this
-example is a coroutine that we start with the [runBlocking] coroutine builder.
-Our main coroutine receives on the flowable using the `source.collect { ... }` expression.
-The main coroutine is _suspended_ while it waits for the source to emit an item.
-When the last items are emitted by `Flowable.range(1, 5)` it
-_resumes_ the main coroutine, which gets dispatched onto the main thread to print this
- last element at a later point in time, while the source completes and prints "Finally".
-
-### Backpressure
-
-Backpressure is one of the most interesting and complex aspects of reactive streams. Coroutines can
-_suspend_ and they provide a natural answer to handling backpressure.
-
-In Rx Java 2.x, the backpressure-capable class is called
-[Flowable](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html).
-In the following example, we use [rxFlowable] coroutine builder from `kotlinx-coroutines-rx2` module to define a
-flowable that sends three integers from 1 to 3.
-It prints a message to the output before invocation of the
-suspending [send][SendChannel.send] function, so that we can study how it operates.
-
-The integers are generated in the context of the main thread, but the subscription is shifted
-to another thread using Rx
-[observeOn](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#observeOn(io.reactivex.Scheduler,%20boolean,%20int))
-operator with a buffer of size 1.
-The subscriber is slow. It takes 500 ms to process each item, which is simulated using `Thread.sleep`.
-
-<!--- INCLUDE
-import io.reactivex.schedulers.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.rx2.*
-import kotlin.coroutines.*
--->
-
-```kotlin
-fun main() = runBlocking<Unit> {
- // coroutine -- fast producer of elements in the context of the main thread
- val source = rxFlowable {
- for (x in 1..3) {
- send(x) // this is a suspending function
- println("Sent $x") // print after successfully sent item
- }
- }
- // subscribe on another thread with a slow subscriber using Rx
- source
- .observeOn(Schedulers.io(), false, 1) // specify buffer size of 1 item
- .doOnComplete { println("Complete") }
- .subscribe { x ->
- Thread.sleep(500) // 500ms to process each item
- println("Processed $x")
- }
- delay(2000) // suspend the main thread for a few seconds
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-basic-05.kt).
-
-The output of this code nicely illustrates how backpressure works with coroutines:
-
-```text
-Sent 1
-Processed 1
-Sent 2
-Processed 2
-Sent 3
-Processed 3
-Complete
-```
-
-<!--- TEST -->
-
-We see here how the producer coroutine puts the first element in the buffer and is suspended while trying to send another
-one. Only after the consumer processes the first item, the producer sends the second one and resumes, etc.
-
-
-### Rx Subject vs BroadcastChannel
-
-RxJava has a concept of [Subject](https://github.com/ReactiveX/RxJava/wiki/Subject) which is an object that
-effectively broadcasts elements to all its subscribers. The matching concept in the coroutines world is called a
-[BroadcastChannel]. There is a variety of subjects in Rx with
-[BehaviorSubject](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/subjects/BehaviorSubject.html) being
-the one used to manage state:
-
-<!--- INCLUDE
-import io.reactivex.subjects.BehaviorSubject
--->
-
-```kotlin
-fun main() {
- val subject = BehaviorSubject.create<String>()
- subject.onNext("one")
- subject.onNext("two") // updates the state of BehaviorSubject, "one" value is lost
- // now subscribe to this subject and print everything
- subject.subscribe(System.out::println)
- subject.onNext("three")
- subject.onNext("four")
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-basic-06.kt).
-
-This code prints the current state of the subject on subscription and all its further updates:
-
-
-```text
-two
-three
-four
-```
-
-<!--- TEST -->
-
-You can subscribe to subjects from a coroutine just as with any other reactive stream:
-
-<!--- INCLUDE
-import io.reactivex.subjects.BehaviorSubject
-import kotlinx.coroutines.*
-import kotlinx.coroutines.rx2.collect
--->
-
-```kotlin
-fun main() = runBlocking<Unit> {
- val subject = BehaviorSubject.create<String>()
- subject.onNext("one")
- subject.onNext("two")
- // now launch a coroutine to print everything
- GlobalScope.launch(Dispatchers.Unconfined) { // launch coroutine in unconfined context
- subject.collect { println(it) }
- }
- subject.onNext("three")
- subject.onNext("four")
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-basic-07.kt).
-
-The result is the same:
-
-```text
-two
-three
-four
-```
-
-<!--- TEST -->
-
-Here we use the [Dispatchers.Unconfined] coroutine context to launch a consuming coroutine with the same behavior as subscription in Rx.
-It basically means that the launched coroutine is going to be immediately executed in the same thread that
-is emitting elements. Contexts are covered in more details in a [separate section](#coroutine-context).
-
-The advantage of coroutines is that it is easy to get conflation behavior for single-threaded UI updates.
-A typical UI application does not need to react to every state change. Only the most recent state is relevant.
-A sequence of back-to-back updates to the application state needs to get reflected in UI only once,
-as soon as the UI thread is free. For the following example we are going to simulate this by launching
-a consuming coroutine in the context of the main thread and use the [yield] function to simulate a break in the
-sequence of updates and to release the main thread:
-
-<!--- INCLUDE
-import io.reactivex.subjects.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.rx2.*
-import kotlin.coroutines.*
--->
-
-```kotlin
-fun main() = runBlocking<Unit> {
- val subject = BehaviorSubject.create<String>()
- subject.onNext("one")
- subject.onNext("two")
- // now launch a coroutine to print the most recent update
- launch { // use the context of the main thread for a coroutine
- subject.collect { println(it) }
- }
- subject.onNext("three")
- subject.onNext("four")
- yield() // yield the main thread to the launched coroutine <--- HERE
- subject.onComplete() // now complete the subject's sequence to cancel the consumer, too
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-basic-08.kt).
-
-Now the coroutine processes (prints) only the most recent update:
-
-```text
-four
-```
-
-<!--- TEST -->
-
-The corresponding behavior in the pure coroutines world is implemented by [ConflatedBroadcastChannel]
-that provides the same logic on top of coroutine channels directly,
-without going through the bridge to the reactive streams:
-
-<!--- INCLUDE
-import kotlinx.coroutines.channels.*
-import kotlinx.coroutines.*
-import kotlin.coroutines.*
--->
-
-```kotlin
-fun main() = runBlocking<Unit> {
- val broadcast = ConflatedBroadcastChannel<String>()
- broadcast.offer("one")
- broadcast.offer("two")
- // now launch a coroutine to print the most recent update
- launch { // use the context of the main thread for a coroutine
- broadcast.consumeEach { println(it) }
- }
- broadcast.offer("three")
- broadcast.offer("four")
- yield() // yield the main thread to the launched coroutine
- broadcast.close() // now close the broadcast channel to cancel the consumer, too
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-basic-09.kt).
-
-It produces the same output as the previous example based on `BehaviorSubject`:
-
-```text
-four
-```
-
-<!--- TEST -->
-
-Another implementation of [BroadcastChannel] is `ArrayBroadcastChannel` with an array-based buffer of
-a specified `capacity`. It can be created with `BroadcastChannel(capacity)`.
-It delivers every event to every
-subscriber as soon as their corresponding subscriptions are opened. It corresponds to
-[PublishSubject](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/subjects/PublishSubject.html) in Rx.
-The capacity of the buffer in the constructor of `ArrayBroadcastChannel` controls the numbers of elements
-that can be sent before the sender is suspended waiting for a receiver to receive those elements.
-
-## Operators
-
-Full-featured reactive stream libraries, like Rx, come with
-[a very large set of operators](https://reactivex.io/documentation/operators.html) to create, transform, combine
-and otherwise process the corresponding streams. Creating your own operators with support for
-back-pressure is [notoriously](https://akarnokd.blogspot.ru/2015/05/pitfalls-of-operator-implementations.html)
-[difficult](https://github.com/ReactiveX/RxJava/wiki/Writing-operators-for-2.0).
-
-Coroutines and channels are designed to provide an opposite experience. There are no built-in operators,
-but processing streams of elements is extremely simple and back-pressure is supported automatically
-without you having to explicitly think about it.
-
-This section shows a coroutine-based implementation of several reactive stream operators.
-
-### Range
-
-Let's roll out own implementation of
-[range](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#range(int,%20int))
-operator for the reactive streams' `Publisher` interface. The asynchronous clean-slate implementation of this operator for
-reactive streams is explained in
-[this blog post](https://akarnokd.blogspot.ru/2017/03/java-9-flow-api-asynchronous-integer.html).
-It takes a lot of code.
-Here is the corresponding code with coroutines:
-
-<!--- INCLUDE
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import kotlin.coroutines.CoroutineContext
--->
-
-```kotlin
-fun CoroutineScope.range(context: CoroutineContext, start: Int, count: Int) = publish<Int>(context) {
- for (x in start until start + count) send(x)
-}
-```
-
-Here, `CoroutineScope` and `context` are used instead of an `Executor` and all the backpressure aspects are taken care
-of by the coroutines machinery. Note that this implementation depends only on the small reactive streams library
-that defines the `Publisher` interface and its friends.
-
-Using it from a coroutine is straightforward:
-
-```kotlin
-fun main() = runBlocking<Unit> {
- // Range inherits parent job from runBlocking, but overrides dispatcher with Dispatchers.Default
- range(Dispatchers.Default, 1, 5).collect { println(it) }
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-operators-01.kt).
-
-The result of this code is quite expected:
-
-```text
-1
-2
-3
-4
-5
-```
-
-<!--- TEST -->
-
-### Fused filter-map hybrid
-
-Reactive operators like
-[filter](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#filter(io.reactivex.functions.Predicate)) and
-[map](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#map(io.reactivex.functions.Function))
-are trivial to implement with coroutines. For a bit of challenge and showcase, let us combine them
-into the single `fusedFilterMap` operator:
-
-<!--- INCLUDE
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import org.reactivestreams.*
-import kotlin.coroutines.*
--->
-
-```kotlin
-fun <T, R> Publisher<T>.fusedFilterMap(
- context: CoroutineContext, // the context to execute this coroutine in
- predicate: (T) -> Boolean, // the filter predicate
- mapper: (T) -> R // the mapper function
-) = publish<R>(context) {
- collect { // collect the source stream
- if (predicate(it)) // filter part
- send(mapper(it)) // map part
- }
-}
-```
-
-Using `range` from the previous example we can test our `fusedFilterMap`
-by filtering for even numbers and mapping them to strings:
-
-<!--- INCLUDE
-
-fun CoroutineScope.range(start: Int, count: Int) = publish<Int> {
- for (x in start until start + count) send(x)
-}
--->
-
-```kotlin
-fun main() = runBlocking<Unit> {
- range(1, 5)
- .fusedFilterMap(Dispatchers.Unconfined, { it % 2 == 0}, { "$it is even" })
- .collect { println(it) } // print all the resulting strings
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-operators-02.kt).
-
-It is not hard to see that the result is going to be:
-
-```text
-2 is even
-4 is even
-```
-
-<!--- TEST -->
-
-### Take until
-
-Let's implement our own version of
-[takeUntil](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#takeUntil(org.reactivestreams.Publisher))
-operator. It is quite [tricky](https://akarnokd.blogspot.ru/2015/05/pitfalls-of-operator-implementations.html)
-as subscriptions to two streams need to be tracked and managed.
-We need to relay all the elements from the source stream until the other stream either completes or
-emits anything. However, we have the [select] expression to rescue us in the coroutines implementation:
-
-<!--- INCLUDE
-import kotlinx.coroutines.channels.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import kotlinx.coroutines.selects.*
-import org.reactivestreams.*
-import kotlin.coroutines.*
--->
-
-```kotlin
-fun <T, U> Publisher<T>.takeUntil(context: CoroutineContext, other: Publisher<U>) = publish<T>(context) {
- this@takeUntil.openSubscription().consume { // explicitly open channel to Publisher<T>
- val current = this
- other.openSubscription().consume { // explicitly open channel to Publisher<U>
- val other = this
- whileSelect {
- other.onReceive { false } // bail out on any received element from `other`
- current.onReceive { send(it); true } // resend element from this channel and continue
- }
- }
- }
-}
-```
-
-This code is using [whileSelect] as a nicer shortcut to `while(select{...}) {}` loop and Kotlin's
-[consume] expressions to close the channels on exit, which unsubscribes from the corresponding publishers.
-
-The following hand-written combination of
-[range](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#range(int,%20int)) with
-[interval](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#interval(long,%20java.util.concurrent.TimeUnit,%20io.reactivex.Scheduler))
-is used for testing. It is coded using a `publish` coroutine builder
-(its pure-Rx implementation is shown in later sections):
-
-```kotlin
-fun CoroutineScope.rangeWithInterval(time: Long, start: Int, count: Int) = publish<Int> {
- for (x in start until start + count) {
- delay(time) // wait before sending each number
- send(x)
- }
-}
-```
-
-The following code shows how `takeUntil` works:
-
-```kotlin
-fun main() = runBlocking<Unit> {
- val slowNums = rangeWithInterval(200, 1, 10) // numbers with 200ms interval
- val stop = rangeWithInterval(500, 1, 10) // the first one after 500ms
- slowNums.takeUntil(Dispatchers.Unconfined, stop).collect { println(it) } // let's test it
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-operators-03.kt).
-
-Producing
-
-```text
-1
-2
-```
-
-<!--- TEST -->
-
-### Merge
-
-There are always at least two ways for processing multiple streams of data with coroutines. One way involving
-[select] was shown in the previous example. The other way is just to launch multiple coroutines. Let
-us implement
-[merge](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#merge(org.reactivestreams.Publisher))
-operator using the latter approach:
-
-<!--- INCLUDE
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import org.reactivestreams.*
-import kotlin.coroutines.*
--->
-
-```kotlin
-fun <T> Publisher<Publisher<T>>.merge(context: CoroutineContext) = publish<T>(context) {
- collect { pub -> // for each publisher collected
- launch { // launch a child coroutine
- pub.collect { send(it) } // resend all element from this publisher
- }
- }
-}
-```
-
-Notice that all the coroutines that are
-being launched here are the children of the `publish`
-coroutine and will get cancelled when the `publish` coroutine is cancelled or is otherwise completed.
-Moreover, since the parent coroutine waits until all the children are complete, this implementation fully
-merges all the received streams.
-
-For a test, let us start with the `rangeWithInterval` function from the previous example and write a
-producer that sends its results twice with some delay:
-
-<!--- INCLUDE
-
-fun CoroutineScope.rangeWithInterval(time: Long, start: Int, count: Int) = publish<Int> {
- for (x in start until start + count) {
- delay(time) // wait before sending each number
- send(x)
- }
-}
--->
-
-```kotlin
-fun CoroutineScope.testPub() = publish<Publisher<Int>> {
- send(rangeWithInterval(250, 1, 4)) // number 1 at 250ms, 2 at 500ms, 3 at 750ms, 4 at 1000ms
- delay(100) // wait for 100 ms
- send(rangeWithInterval(500, 11, 3)) // number 11 at 600ms, 12 at 1100ms, 13 at 1600ms
- delay(1100) // wait for 1.1s - done in 1.2 sec after start
-}
-```
-
-The test code is to use `merge` on `testPub` and to display the results:
-
-```kotlin
-fun main() = runBlocking<Unit> {
- testPub().merge(Dispatchers.Unconfined).collect { println(it) } // print the whole stream
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-operators-04.kt).
-
-And the results should be:
-
-```text
-1
-2
-11
-3
-4
-12
-13
-```
-
-<!--- TEST -->
-
-## Coroutine context
-
-All the example operators that are shown in the previous section have an explicit
-[CoroutineContext](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines/-coroutine-context/)
-parameter. In the Rx world it roughly corresponds to
-a [Scheduler](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Scheduler.html).
-
-### Threads with Rx
-
-The following example shows the basics of threading context management with Rx.
-Here `rangeWithIntervalRx` is an implementation of `rangeWithInterval` function using Rx
-`zip`, `range`, and `interval` operators.
-
-<!--- INCLUDE
-import io.reactivex.*
-import io.reactivex.functions.BiFunction
-import io.reactivex.schedulers.Schedulers
-import java.util.concurrent.TimeUnit
--->
-
-```kotlin
-fun rangeWithIntervalRx(scheduler: Scheduler, time: Long, start: Int, count: Int): Flowable<Int> =
- Flowable.zip(
- Flowable.range(start, count),
- Flowable.interval(time, TimeUnit.MILLISECONDS, scheduler),
- BiFunction { x, _ -> x })
-
-fun main() {
- rangeWithIntervalRx(Schedulers.computation(), 100, 1, 3)
- .subscribe { println("$it on thread ${Thread.currentThread().name}") }
- Thread.sleep(1000)
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-context-01.kt).
-
-We are explicitly passing the
-[Schedulers.computation()](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/schedulers/Schedulers.html#computation())
-scheduler to our `rangeWithIntervalRx` operator and
-it is going to be executed in Rx computation thread pool. The output is going to be similar to the following one:
-
-```text
-1 on thread RxComputationThreadPool-1
-2 on thread RxComputationThreadPool-1
-3 on thread RxComputationThreadPool-1
-```
-
-<!--- TEST FLEXIBLE_THREAD -->
-
-### Threads with coroutines
-
-In the world of coroutines `Schedulers.computation()` roughly corresponds to [Dispatchers.Default],
-so the previous example is similar to the following one:
-
-<!--- INCLUDE
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import kotlin.coroutines.CoroutineContext
--->
-
-```kotlin
-fun rangeWithInterval(context: CoroutineContext, time: Long, start: Int, count: Int) = publish<Int>(context) {
- for (x in start until start + count) {
- delay(time) // wait before sending each number
- send(x)
- }
-}
-
-fun main() {
- Flowable.fromPublisher(rangeWithInterval(Dispatchers.Default, 100, 1, 3))
- .subscribe { println("$it on thread ${Thread.currentThread().name}") }
- Thread.sleep(1000)
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-context-02.kt).
-
-The produced output is going to be similar to:
-
-```text
-1 on thread ForkJoinPool.commonPool-worker-1
-2 on thread ForkJoinPool.commonPool-worker-1
-3 on thread ForkJoinPool.commonPool-worker-1
-```
-
-<!--- TEST LINES_START -->
-
-Here we've used Rx
-[subscribe](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#subscribe(io.reactivex.functions.Consumer))
-operator that does not have its own scheduler and operates on the same thread that the publisher -- on a default
-shared pool of threads in this example.
-
-### Rx observeOn
-
-In Rx you use special operators to modify the threading context for operations in the chain. You
-can find some [good guides](https://tomstechnicalblog.blogspot.ru/2016/02/rxjava-understanding-observeon-and.html)
-about them, if you are not familiar.
-
-For example, there is
-[observeOn](https://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#observeOn(io.reactivex.Scheduler))
-operator. Let us modify the previous example to observe using `Schedulers.computation()`:
-
-<!--- INCLUDE
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import io.reactivex.schedulers.Schedulers
-import kotlin.coroutines.CoroutineContext
--->
-
-```kotlin
-fun rangeWithInterval(context: CoroutineContext, time: Long, start: Int, count: Int) = publish<Int>(context) {
- for (x in start until start + count) {
- delay(time) // wait before sending each number
- send(x)
- }
-}
-
-fun main() {
- Flowable.fromPublisher(rangeWithInterval(Dispatchers.Default, 100, 1, 3))
- .observeOn(Schedulers.computation()) // <-- THIS LINE IS ADDED
- .subscribe { println("$it on thread ${Thread.currentThread().name}") }
- Thread.sleep(1000)
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-context-03.kt).
-
-Here is the difference in output, notice "RxComputationThreadPool":
-
-```text
-1 on thread RxComputationThreadPool-1
-2 on thread RxComputationThreadPool-1
-3 on thread RxComputationThreadPool-1
-```
-
-<!--- TEST FLEXIBLE_THREAD -->
-
-### Coroutine context to rule them all
-
-A coroutine is always working in some context. For example, let us start a coroutine
-in the main thread with [runBlocking] and iterate over the result of the Rx version of `rangeWithIntervalRx` operator,
-instead of using Rx `subscribe` operator:
-
-<!--- INCLUDE
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import io.reactivex.functions.BiFunction
-import io.reactivex.schedulers.Schedulers
-import java.util.concurrent.TimeUnit
--->
-
-```kotlin
-fun rangeWithIntervalRx(scheduler: Scheduler, time: Long, start: Int, count: Int): Flowable<Int> =
- Flowable.zip(
- Flowable.range(start, count),
- Flowable.interval(time, TimeUnit.MILLISECONDS, scheduler),
- BiFunction { x, _ -> x })
-
-fun main() = runBlocking<Unit> {
- rangeWithIntervalRx(Schedulers.computation(), 100, 1, 3)
- .collect { println("$it on thread ${Thread.currentThread().name}") }
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-context-04.kt).
-
-The resulting messages are going to be printed in the main thread:
-
-```text
-1 on thread main
-2 on thread main
-3 on thread main
-```
-
-<!--- TEST LINES_START -->
-
-### Unconfined context
-
-Most Rx operators do not have any specific thread (scheduler) associated with them and are working
-in whatever thread they happen to be invoked. We've seen it in the example with the `subscribe` operator
-in the [threads with Rx](#threads-with-rx) section.
-
-In the world of coroutines, [Dispatchers.Unconfined] context serves a similar role. Let us modify our previous example,
-but instead of iterating over the source `Flowable` from the `runBlocking` coroutine that is confined
-to the main thread, we launch a new coroutine in the `Dispatchers.Unconfined` context, while the main coroutine
-simply waits for its completion using [Job.join]:
-
-<!--- INCLUDE
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import io.reactivex.functions.BiFunction
-import io.reactivex.schedulers.Schedulers
-import java.util.concurrent.TimeUnit
--->
-
-```kotlin
-fun rangeWithIntervalRx(scheduler: Scheduler, time: Long, start: Int, count: Int): Flowable<Int> =
- Flowable.zip(
- Flowable.range(start, count),
- Flowable.interval(time, TimeUnit.MILLISECONDS, scheduler),
- BiFunction { x, _ -> x })
-
-fun main() = runBlocking<Unit> {
- val job = launch(Dispatchers.Unconfined) { // launch a new coroutine in Unconfined context (without its own thread pool)
- rangeWithIntervalRx(Schedulers.computation(), 100, 1, 3)
- .collect { println("$it on thread ${Thread.currentThread().name}") }
- }
- job.join() // wait for our coroutine to complete
-}
-```
-
-> You can get full code [here](kotlinx-coroutines-rx2/test/guide/example-reactive-context-05.kt).
-
-Now, the output shows that the code of the coroutine is executing in the Rx computation thread pool, just
-like our initial example using the Rx `subscribe` operator.
-
-```text
-1 on thread RxComputationThreadPool-1
-2 on thread RxComputationThreadPool-1
-3 on thread RxComputationThreadPool-1
-```
-
-<!--- TEST LINES_START -->
-
-Note that the [Dispatchers.Unconfined] context should be used with care. It may improve the overall performance on certain tests,
-due to the increased stack-locality of operations and less scheduling overhead, but it also produces deeper stacks
-and makes it harder to reason about asynchronicity of the code that is using it.
-
-If a coroutine sends an element to a channel, then the thread that invoked the
-[send][SendChannel.send] may start executing the code of the coroutine with the [Dispatchers.Unconfined] dispatcher.
-The original producer coroutine that invoked `send` is paused until the unconfined consumer coroutine hits its next
-suspension point. This is very similar to a lock-step single-threaded `onNext` execution in the Rx world in the absense
-of thread-shifting operators. It is a normal default for Rx, because operators are usually doing very small chunks
-of work and you have to combine many operators for a complex processing. However, this is unusual with coroutines,
-where you can have an arbitrary complex processing in a coroutine. Usually, you only need to chain stream-processing
-coroutines for complex pipelines with fan-in and fan-out between multiple worker coroutines.
-
-<!--- MODULE kotlinx-coroutines-core -->
-<!--- INDEX kotlinx.coroutines -->
-[runBlocking]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/run-blocking.html
-[Dispatchers.Unconfined]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-dispatchers/-unconfined.html
-[yield]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/yield.html
-[Dispatchers.Default]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-dispatchers/-default.html
-[Job.join]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job/join.html
-<!--- INDEX kotlinx.coroutines.channels -->
-[Channel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-channel/index.html
-[ReceiveChannel.receive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-receive-channel/receive.html
-[produce]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/produce.html
-[consumeEach]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/consume-each.html
-[ReceiveChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-receive-channel/index.html
-[ReceiveChannel.cancel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-receive-channel/cancel.html
-[consume]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/consume.html
-[SendChannel.send]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-send-channel/send.html
-[BroadcastChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-broadcast-channel/index.html
-[ConflatedBroadcastChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-conflated-broadcast-channel/index.html
-<!--- INDEX kotlinx.coroutines.selects -->
-[select]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.selects/select.html
-[whileSelect]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.selects/while-select.html
-<!--- MODULE kotlinx-coroutines-reactive -->
-<!--- INDEX kotlinx.coroutines.reactive -->
-[publish]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.reactive/publish.html
-[org.reactivestreams.Publisher.collect]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.reactive/org.reactivestreams.-publisher/collect.html
-[org.reactivestreams.Publisher.openSubscription]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.reactive/org.reactivestreams.-publisher/open-subscription.html
-<!--- MODULE kotlinx-coroutines-rx2 -->
-<!--- INDEX kotlinx.coroutines.rx2 -->
-[rxFlowable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.rx2/rx-flowable.html
-<!--- END -->
-
-
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-01.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-01.kt
deleted file mode 100644
index f3bc344..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-01.kt
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.basic01
-
-import kotlinx.coroutines.*
-import kotlinx.coroutines.channels.*
-import kotlin.coroutines.*
-
-fun main() = runBlocking<Unit> {
- // create a channel that produces numbers from 1 to 3 with 200ms delays between them
- val source = produce<Int> {
- println("Begin") // mark the beginning of this coroutine in output
- for (x in 1..3) {
- delay(200) // wait for 200ms
- send(x) // send number x to the channel
- }
- }
- // print elements from the source
- println("Elements:")
- source.consumeEach { // consume elements from it
- println(it)
- }
- // print elements from the source AGAIN
- println("Again:")
- source.consumeEach { // consume elements from it
- println(it)
- }
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-02.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-02.kt
deleted file mode 100644
index 0e0ff2e..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-02.kt
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.basic02
-
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import kotlin.coroutines.*
-
-fun main() = runBlocking<Unit> {
- // create a publisher that produces numbers from 1 to 3 with 200ms delays between them
- val source = publish<Int> {
- // ^^^^^^^ <--- Difference from the previous examples is here
- println("Begin") // mark the beginning of this coroutine in output
- for (x in 1..3) {
- delay(200) // wait for 200ms
- send(x) // send number x to the channel
- }
- }
- // print elements from the source
- println("Elements:")
- source.collect { // collect elements from it
- println(it)
- }
- // print elements from the source AGAIN
- println("Again:")
- source.collect { // collect elements from it
- println(it)
- }
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-03.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-03.kt
deleted file mode 100644
index b84fc08..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-03.kt
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.basic03
-
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.channels.*
-import kotlinx.coroutines.reactive.*
-
-fun main() = runBlocking<Unit> {
- val source = Flowable.range(1, 5) // a range of five numbers
- .doOnSubscribe { println("OnSubscribe") } // provide some insight
- .doOnComplete { println("OnComplete") } // ...
- .doFinally { println("Finally") } // ... into what's going on
- var cnt = 0
- source.openSubscription().consume { // open channel to the source
- for (x in this) { // iterate over the channel to receive elements from it
- println(x)
- if (++cnt >= 3) break // break when 3 elements are printed
- }
- // Note: `consume` cancels the channel when this block of code is complete
- }
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-04.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-04.kt
deleted file mode 100644
index a08c41f..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-04.kt
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.basic04
-
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import kotlin.coroutines.*
-
-fun main() = runBlocking<Unit> {
- val source = Flowable.range(1, 5) // a range of five numbers
- .doOnSubscribe { println("OnSubscribe") } // provide some insight
- .doOnComplete { println("OnComplete") } // ...
- .doFinally { println("Finally") } // ... into what's going on
- // collect the source fully
- source.collect { println(it) }
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-05.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-05.kt
deleted file mode 100644
index e6428b9..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-05.kt
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.basic05
-
-import io.reactivex.schedulers.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.rx2.*
-import kotlin.coroutines.*
-
-fun main() = runBlocking<Unit> {
- // coroutine -- fast producer of elements in the context of the main thread
- val source = rxFlowable {
- for (x in 1..3) {
- send(x) // this is a suspending function
- println("Sent $x") // print after successfully sent item
- }
- }
- // subscribe on another thread with a slow subscriber using Rx
- source
- .observeOn(Schedulers.io(), false, 1) // specify buffer size of 1 item
- .doOnComplete { println("Complete") }
- .subscribe { x ->
- Thread.sleep(500) // 500ms to process each item
- println("Processed $x")
- }
- delay(2000) // suspend the main thread for a few seconds
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-06.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-06.kt
deleted file mode 100644
index 1f3747f..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-06.kt
+++ /dev/null
@@ -1,18 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.basic06
-
-import io.reactivex.subjects.BehaviorSubject
-
-fun main() {
- val subject = BehaviorSubject.create<String>()
- subject.onNext("one")
- subject.onNext("two") // updates the state of BehaviorSubject, "one" value is lost
- // now subscribe to this subject and print everything
- subject.subscribe(System.out::println)
- subject.onNext("three")
- subject.onNext("four")
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-07.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-07.kt
deleted file mode 100644
index b4cc9fc..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-07.kt
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.basic07
-
-import io.reactivex.subjects.BehaviorSubject
-import kotlinx.coroutines.*
-import kotlinx.coroutines.rx2.collect
-
-fun main() = runBlocking<Unit> {
- val subject = BehaviorSubject.create<String>()
- subject.onNext("one")
- subject.onNext("two")
- // now launch a coroutine to print everything
- GlobalScope.launch(Dispatchers.Unconfined) { // launch coroutine in unconfined context
- subject.collect { println(it) }
- }
- subject.onNext("three")
- subject.onNext("four")
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-08.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-08.kt
deleted file mode 100644
index 8e17ac9..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-08.kt
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.basic08
-
-import io.reactivex.subjects.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.rx2.*
-import kotlin.coroutines.*
-
-fun main() = runBlocking<Unit> {
- val subject = BehaviorSubject.create<String>()
- subject.onNext("one")
- subject.onNext("two")
- // now launch a coroutine to print the most recent update
- launch { // use the context of the main thread for a coroutine
- subject.collect { println(it) }
- }
- subject.onNext("three")
- subject.onNext("four")
- yield() // yield the main thread to the launched coroutine <--- HERE
- subject.onComplete() // now complete the subject's sequence to cancel the consumer, too
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-09.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-09.kt
deleted file mode 100644
index 738c4ab..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-basic-09.kt
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.basic09
-
-import kotlinx.coroutines.channels.*
-import kotlinx.coroutines.*
-import kotlin.coroutines.*
-
-fun main() = runBlocking<Unit> {
- val broadcast = ConflatedBroadcastChannel<String>()
- broadcast.offer("one")
- broadcast.offer("two")
- // now launch a coroutine to print the most recent update
- launch { // use the context of the main thread for a coroutine
- broadcast.consumeEach { println(it) }
- }
- broadcast.offer("three")
- broadcast.offer("four")
- yield() // yield the main thread to the launched coroutine
- broadcast.close() // now close the broadcast channel to cancel the consumer, too
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-01.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-01.kt
deleted file mode 100644
index b12e92a..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-01.kt
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.context01
-
-import io.reactivex.*
-import io.reactivex.functions.BiFunction
-import io.reactivex.schedulers.Schedulers
-import java.util.concurrent.TimeUnit
-
-fun rangeWithIntervalRx(scheduler: Scheduler, time: Long, start: Int, count: Int): Flowable<Int> =
- Flowable.zip(
- Flowable.range(start, count),
- Flowable.interval(time, TimeUnit.MILLISECONDS, scheduler),
- BiFunction { x, _ -> x })
-
-fun main() {
- rangeWithIntervalRx(Schedulers.computation(), 100, 1, 3)
- .subscribe { println("$it on thread ${Thread.currentThread().name}") }
- Thread.sleep(1000)
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-02.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-02.kt
deleted file mode 100644
index b87849a..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-02.kt
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.context02
-
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import kotlin.coroutines.CoroutineContext
-
-fun rangeWithInterval(context: CoroutineContext, time: Long, start: Int, count: Int) = publish<Int>(context) {
- for (x in start until start + count) {
- delay(time) // wait before sending each number
- send(x)
- }
-}
-
-fun main() {
- Flowable.fromPublisher(rangeWithInterval(Dispatchers.Default, 100, 1, 3))
- .subscribe { println("$it on thread ${Thread.currentThread().name}") }
- Thread.sleep(1000)
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-03.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-03.kt
deleted file mode 100644
index 1a214ce..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-03.kt
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.context03
-
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import io.reactivex.schedulers.Schedulers
-import kotlin.coroutines.CoroutineContext
-
-fun rangeWithInterval(context: CoroutineContext, time: Long, start: Int, count: Int) = publish<Int>(context) {
- for (x in start until start + count) {
- delay(time) // wait before sending each number
- send(x)
- }
-}
-
-fun main() {
- Flowable.fromPublisher(rangeWithInterval(Dispatchers.Default, 100, 1, 3))
- .observeOn(Schedulers.computation()) // <-- THIS LINE IS ADDED
- .subscribe { println("$it on thread ${Thread.currentThread().name}") }
- Thread.sleep(1000)
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-04.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-04.kt
deleted file mode 100644
index 3c5d3fb..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-04.kt
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.context04
-
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import io.reactivex.functions.BiFunction
-import io.reactivex.schedulers.Schedulers
-import java.util.concurrent.TimeUnit
-
-fun rangeWithIntervalRx(scheduler: Scheduler, time: Long, start: Int, count: Int): Flowable<Int> =
- Flowable.zip(
- Flowable.range(start, count),
- Flowable.interval(time, TimeUnit.MILLISECONDS, scheduler),
- BiFunction { x, _ -> x })
-
-fun main() = runBlocking<Unit> {
- rangeWithIntervalRx(Schedulers.computation(), 100, 1, 3)
- .collect { println("$it on thread ${Thread.currentThread().name}") }
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-05.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-05.kt
deleted file mode 100644
index 61b54b2..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-context-05.kt
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.context05
-
-import io.reactivex.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import io.reactivex.functions.BiFunction
-import io.reactivex.schedulers.Schedulers
-import java.util.concurrent.TimeUnit
-
-fun rangeWithIntervalRx(scheduler: Scheduler, time: Long, start: Int, count: Int): Flowable<Int> =
- Flowable.zip(
- Flowable.range(start, count),
- Flowable.interval(time, TimeUnit.MILLISECONDS, scheduler),
- BiFunction { x, _ -> x })
-
-fun main() = runBlocking<Unit> {
- val job = launch(Dispatchers.Unconfined) { // launch a new coroutine in Unconfined context (without its own thread pool)
- rangeWithIntervalRx(Schedulers.computation(), 100, 1, 3)
- .collect { println("$it on thread ${Thread.currentThread().name}") }
- }
- job.join() // wait for our coroutine to complete
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-01.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-01.kt
deleted file mode 100644
index 8268ef2..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-01.kt
+++ /dev/null
@@ -1,19 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.operators01
-
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import kotlin.coroutines.CoroutineContext
-
-fun CoroutineScope.range(context: CoroutineContext, start: Int, count: Int) = publish<Int>(context) {
- for (x in start until start + count) send(x)
-}
-
-fun main() = runBlocking<Unit> {
- // Range inherits parent job from runBlocking, but overrides dispatcher with Dispatchers.Default
- range(Dispatchers.Default, 1, 5).collect { println(it) }
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-02.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-02.kt
deleted file mode 100644
index 5f07ba4..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-02.kt
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.operators02
-
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import org.reactivestreams.*
-import kotlin.coroutines.*
-
-fun <T, R> Publisher<T>.fusedFilterMap(
- context: CoroutineContext, // the context to execute this coroutine in
- predicate: (T) -> Boolean, // the filter predicate
- mapper: (T) -> R // the mapper function
-) = publish<R>(context) {
- collect { // collect the source stream
- if (predicate(it)) // filter part
- send(mapper(it)) // map part
- }
-}
-
-fun CoroutineScope.range(start: Int, count: Int) = publish<Int> {
- for (x in start until start + count) send(x)
-}
-
-fun main() = runBlocking<Unit> {
- range(1, 5)
- .fusedFilterMap(Dispatchers.Unconfined, { it % 2 == 0}, { "$it is even" })
- .collect { println(it) } // print all the resulting strings
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-03.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-03.kt
deleted file mode 100644
index 818a792..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-03.kt
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.operators03
-
-import kotlinx.coroutines.channels.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import kotlinx.coroutines.selects.*
-import org.reactivestreams.*
-import kotlin.coroutines.*
-
-fun <T, U> Publisher<T>.takeUntil(context: CoroutineContext, other: Publisher<U>) = publish<T>(context) {
- this@takeUntil.openSubscription().consume { // explicitly open channel to Publisher<T>
- val current = this
- other.openSubscription().consume { // explicitly open channel to Publisher<U>
- val other = this
- whileSelect {
- other.onReceive { false } // bail out on any received element from `other`
- current.onReceive { send(it); true } // resend element from this channel and continue
- }
- }
- }
-}
-
-fun CoroutineScope.rangeWithInterval(time: Long, start: Int, count: Int) = publish<Int> {
- for (x in start until start + count) {
- delay(time) // wait before sending each number
- send(x)
- }
-}
-
-fun main() = runBlocking<Unit> {
- val slowNums = rangeWithInterval(200, 1, 10) // numbers with 200ms interval
- val stop = rangeWithInterval(500, 1, 10) // the first one after 500ms
- slowNums.takeUntil(Dispatchers.Unconfined, stop).collect { println(it) } // let's test it
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-04.kt b/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-04.kt
deleted file mode 100644
index 12d9c1f..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/example-reactive-operators-04.kt
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.operators04
-
-import kotlinx.coroutines.*
-import kotlinx.coroutines.reactive.*
-import org.reactivestreams.*
-import kotlin.coroutines.*
-
-fun <T> Publisher<Publisher<T>>.merge(context: CoroutineContext) = publish<T>(context) {
- collect { pub -> // for each publisher collected
- launch { // launch a child coroutine
- pub.collect { send(it) } // resend all element from this publisher
- }
- }
-}
-
-fun CoroutineScope.rangeWithInterval(time: Long, start: Int, count: Int) = publish<Int> {
- for (x in start until start + count) {
- delay(time) // wait before sending each number
- send(x)
- }
-}
-
-fun CoroutineScope.testPub() = publish<Publisher<Int>> {
- send(rangeWithInterval(250, 1, 4)) // number 1 at 250ms, 2 at 500ms, 3 at 750ms, 4 at 1000ms
- delay(100) // wait for 100 ms
- send(rangeWithInterval(500, 11, 3)) // number 11 at 600ms, 12 at 1100ms, 13 at 1600ms
- delay(1100) // wait for 1.1s - done in 1.2 sec after start
-}
-
-fun main() = runBlocking<Unit> {
- testPub().merge(Dispatchers.Unconfined).collect { println(it) } // print the whole stream
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/test/GuideReactiveTest.kt b/reactive/kotlinx-coroutines-rx2/test/guide/test/GuideReactiveTest.kt
deleted file mode 100644
index cebfc7b..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/test/GuideReactiveTest.kt
+++ /dev/null
@@ -1,191 +0,0 @@
-// This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
-package kotlinx.coroutines.rx2.guide.test
-
-import kotlinx.coroutines.guide.test.*
-import org.junit.Test
-
-class GuideReactiveTest : ReactiveTestBase() {
-
- @Test
- fun testKotlinxCoroutinesRx2GuideBasic01() {
- test("KotlinxCoroutinesRx2GuideBasic01") { kotlinx.coroutines.rx2.guide.basic01.main() }.verifyLines(
- "Elements:",
- "Begin",
- "1",
- "2",
- "3",
- "Again:"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideBasic02() {
- test("KotlinxCoroutinesRx2GuideBasic02") { kotlinx.coroutines.rx2.guide.basic02.main() }.verifyLines(
- "Elements:",
- "Begin",
- "1",
- "2",
- "3",
- "Again:",
- "Begin",
- "1",
- "2",
- "3"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideBasic03() {
- test("KotlinxCoroutinesRx2GuideBasic03") { kotlinx.coroutines.rx2.guide.basic03.main() }.verifyLines(
- "OnSubscribe",
- "1",
- "2",
- "3",
- "Finally"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideBasic04() {
- test("KotlinxCoroutinesRx2GuideBasic04") { kotlinx.coroutines.rx2.guide.basic04.main() }.verifyLines(
- "OnSubscribe",
- "1",
- "2",
- "3",
- "OnComplete",
- "Finally",
- "4",
- "5"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideBasic05() {
- test("KotlinxCoroutinesRx2GuideBasic05") { kotlinx.coroutines.rx2.guide.basic05.main() }.verifyLines(
- "Sent 1",
- "Processed 1",
- "Sent 2",
- "Processed 2",
- "Sent 3",
- "Processed 3",
- "Complete"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideBasic06() {
- test("KotlinxCoroutinesRx2GuideBasic06") { kotlinx.coroutines.rx2.guide.basic06.main() }.verifyLines(
- "two",
- "three",
- "four"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideBasic07() {
- test("KotlinxCoroutinesRx2GuideBasic07") { kotlinx.coroutines.rx2.guide.basic07.main() }.verifyLines(
- "two",
- "three",
- "four"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideBasic08() {
- test("KotlinxCoroutinesRx2GuideBasic08") { kotlinx.coroutines.rx2.guide.basic08.main() }.verifyLines(
- "four"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideBasic09() {
- test("KotlinxCoroutinesRx2GuideBasic09") { kotlinx.coroutines.rx2.guide.basic09.main() }.verifyLines(
- "four"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideOperators01() {
- test("KotlinxCoroutinesRx2GuideOperators01") { kotlinx.coroutines.rx2.guide.operators01.main() }.verifyLines(
- "1",
- "2",
- "3",
- "4",
- "5"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideOperators02() {
- test("KotlinxCoroutinesRx2GuideOperators02") { kotlinx.coroutines.rx2.guide.operators02.main() }.verifyLines(
- "2 is even",
- "4 is even"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideOperators03() {
- test("KotlinxCoroutinesRx2GuideOperators03") { kotlinx.coroutines.rx2.guide.operators03.main() }.verifyLines(
- "1",
- "2"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideOperators04() {
- test("KotlinxCoroutinesRx2GuideOperators04") { kotlinx.coroutines.rx2.guide.operators04.main() }.verifyLines(
- "1",
- "2",
- "11",
- "3",
- "4",
- "12",
- "13"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideContext01() {
- test("KotlinxCoroutinesRx2GuideContext01") { kotlinx.coroutines.rx2.guide.context01.main() }.verifyLinesFlexibleThread(
- "1 on thread RxComputationThreadPool-1",
- "2 on thread RxComputationThreadPool-1",
- "3 on thread RxComputationThreadPool-1"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideContext02() {
- test("KotlinxCoroutinesRx2GuideContext02") { kotlinx.coroutines.rx2.guide.context02.main() }.verifyLinesStart(
- "1 on thread ForkJoinPool.commonPool-worker-1",
- "2 on thread ForkJoinPool.commonPool-worker-1",
- "3 on thread ForkJoinPool.commonPool-worker-1"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideContext03() {
- test("KotlinxCoroutinesRx2GuideContext03") { kotlinx.coroutines.rx2.guide.context03.main() }.verifyLinesFlexibleThread(
- "1 on thread RxComputationThreadPool-1",
- "2 on thread RxComputationThreadPool-1",
- "3 on thread RxComputationThreadPool-1"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideContext04() {
- test("KotlinxCoroutinesRx2GuideContext04") { kotlinx.coroutines.rx2.guide.context04.main() }.verifyLinesStart(
- "1 on thread main",
- "2 on thread main",
- "3 on thread main"
- )
- }
-
- @Test
- fun testKotlinxCoroutinesRx2GuideContext05() {
- test("KotlinxCoroutinesRx2GuideContext05") { kotlinx.coroutines.rx2.guide.context05.main() }.verifyLinesStart(
- "1 on thread RxComputationThreadPool-1",
- "2 on thread RxComputationThreadPool-1",
- "3 on thread RxComputationThreadPool-1"
- )
- }
-}
diff --git a/reactive/kotlinx-coroutines-rx2/test/guide/test/ReactiveTestBase.kt b/reactive/kotlinx-coroutines-rx2/test/guide/test/ReactiveTestBase.kt
deleted file mode 100644
index 3bedb62..0000000
--- a/reactive/kotlinx-coroutines-rx2/test/guide/test/ReactiveTestBase.kt
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
- */
-
-package kotlinx.coroutines.rx2.guide.test
-
-import io.reactivex.*
-import io.reactivex.disposables.*
-import io.reactivex.plugins.*
-import kotlinx.coroutines.*
-import kotlinx.coroutines.guide.test.*
-import org.junit.*
-import java.util.concurrent.*
-
-open class ReactiveTestBase {
- @Before
- fun setup() {
- RxJavaPlugins.setIoSchedulerHandler(Handler)
- RxJavaPlugins.setComputationSchedulerHandler(Handler)
- ignoreLostThreads(
- "RxComputationThreadPool-",
- "RxCachedThreadScheduler-",
- "RxCachedWorkerPoolEvictor-",
- "RxSchedulerPurge-")
- }
-
- @After
- fun tearDown() {
- RxJavaPlugins.setIoSchedulerHandler(null)
- RxJavaPlugins.setComputationSchedulerHandler(null)
- }
-}
-
-private object Handler : io.reactivex.functions.Function<Scheduler, Scheduler> {
- override fun apply(t: Scheduler): Scheduler = WrapperScheduler(t)
-}
-
-private class WrapperScheduler(private val scheduler: Scheduler) : Scheduler() {
- override fun createWorker(): Worker = WrapperWorker(scheduler.createWorker())
-}
-
-private class WrapperWorker(private val worker: Scheduler.Worker) : Scheduler.Worker() {
- override fun isDisposed(): Boolean = worker.isDisposed
- override fun dispose() = worker.dispose()
- override fun schedule(run: Runnable, delay: Long, unit: TimeUnit): Disposable =
- worker.schedule(wrapTask(run), delay, unit)
-}