Reword some docs in jdk9 and reactive, refactor

* Fixed the broken links in the docs
* Some new links are added to the docs
* More detailed descriptions of error handling for some methods
* Small grammar fixes
* Code formatting changes

Author: dkhalanskyjb

reactive/kotlinx-coroutines-jdk9/src/Await.kt

@@ -4,78 +4,82 @@
 
 package kotlinx.coroutines.jdk9
 
+import kotlinx.coroutines.Job
 import java.util.concurrent.*
 import org.reactivestreams.FlowAdapters
 import kotlinx.coroutines.reactive.*
 
 /**
- * Awaits for the first value from the given publisher without blocking a thread and
- * returns the resulting value or throws the corresponding exception if this publisher had produced error.
+ * Awaits the first value from the given publisher without blocking the thread and returns the resulting value, or, if
+ * the publisher has produced an error, throws the corresponding exception.
  *
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
- * immediately resumes with [CancellationException].
+ * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
+ * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
  *
- * @throws NoSuchElementException if publisher does not emit any value
+ * @throws NoSuchElementException if the publisher does not emit any value
  */
-public suspend fun <T> Flow.Publisher<T>.awaitFirst(): T = FlowAdapters.toPublisher(this).awaitFirst()
+public suspend fun <T> Flow.Publisher<T>.awaitFirst(): T =
+    FlowAdapters.toPublisher(this).awaitFirst()
 
 /**
- * Awaits for the first value from the given observable or the [default] value if none is emitted without blocking a
- * thread and returns the resulting value or throws the corresponding exception if this observable had produced error.
+ * Awaits the first value from the given observable, or returns the [default] value if none is emitted, without blocking
+ * the thread, and returns the resulting value, or, if this observable has produced an error, throws the corresponding
+ * exception.
  *
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
- * immediately resumes with [CancellationException].
+ * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
+ * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
  */
 public suspend fun <T> Flow.Publisher<T>.awaitFirstOrDefault(default: T): T =
-        FlowAdapters.toPublisher(this).awaitFirstOrDefault(default)
+    FlowAdapters.toPublisher(this).awaitFirstOrDefault(default)
 
 /**
- * Awaits for the first value from the given observable or `null` value if none is emitted without blocking a
- * thread and returns the resulting value or throws the corresponding exception if this observable had produced error.
+ * Awaits the first value from the given observable, or returns `null` if none is emitted, without blocking the thread,
+ * and returns the resulting value, or, if this observable has produced an error, throws the corresponding exception.
  *
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
- * immediately resumes with [CancellationException].
+ * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
+ * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
  */
 public suspend fun <T> Flow.Publisher<T>.awaitFirstOrNull(): T? =
-        FlowAdapters.toPublisher(this).awaitFirstOrNull()
+    FlowAdapters.toPublisher(this).awaitFirstOrNull()
 
 /**
- * Awaits for the first value from the given observable or call [defaultValue] to get a value if none is emitted without blocking a
- * thread and returns the resulting value or throws the corresponding exception if this observable had produced error.
+ * Awaits the first value from the given observable, or calls [defaultValue] to get a value if none is emitted, without
+ * blocking the thread, and returns the resulting value, or, if this observable has produced an error, throws the
+ * corresponding exception.
  *
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
- * immediately resumes with [CancellationException].
+ * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
+ * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
  */
 public suspend fun <T> Flow.Publisher<T>.awaitFirstOrElse(defaultValue: () -> T): T =
-        FlowAdapters.toPublisher(this).awaitFirstOrElse(defaultValue)
+    FlowAdapters.toPublisher(this).awaitFirstOrElse(defaultValue)
 
 /**
- * Awaits for the last value from the given publisher without blocking a thread and
- * returns the resulting value or throws the corresponding exception if this publisher had produced error.
+ * Awaits the last value from the given publisher without blocking the thread and
+ * returns the resulting value, or, if this publisher has produced an error, throws the corresponding exception.
  *
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
- * immediately resumes with [CancellationException].
+ * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
+ * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
  *
- * @throws NoSuchElementException if publisher does not emit any value
+ * @throws NoSuchElementException if the publisher does not emit any value
  */
 public suspend fun <T> Flow.Publisher<T>.awaitLast(): T =
-        FlowAdapters.toPublisher(this).awaitLast()
+    FlowAdapters.toPublisher(this).awaitLast()
 
 /**
- * Awaits for the single value from the given publisher without blocking a thread and
- * returns the resulting value or throws the corresponding exception if this publisher had produced error.
+ * Awaits the single value from the given publisher without blocking the thread and returns the resulting value, or,
+ * if this publisher has produced an error, throws the corresponding exception.
  *
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
- * immediately resumes with [CancellationException].
+ * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
+ * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
  *
- * @throws NoSuchElementException if publisher does not emit any value
- * @throws IllegalArgumentException if publisher emits more than one value
+ * @throws NoSuchElementException if the publisher does not emit any value
+ * @throws IllegalArgumentException if the publisher emits more than one value
  */
 public suspend fun <T> Flow.Publisher<T>.awaitSingle(): T =
-        FlowAdapters.toPublisher(this).awaitSingle()
+    FlowAdapters.toPublisher(this).awaitSingle()

reactive/kotlinx-coroutines-jdk9/src/Publish.kt

@@ -6,33 +6,34 @@ package kotlinx.coroutines.jdk9
 
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
+import kotlinx.coroutines.reactive.*
 import java.util.concurrent.*
 import kotlin.coroutines.*
 import org.reactivestreams.FlowAdapters
 
 /**
- * Creates cold reactive [Flow.Publisher] that runs a given [block] in a coroutine.
+ * Creates a cold reactive [Flow.Publisher] that runs a given [block] in a coroutine.
+ *
  * Every time the returned flux is subscribed, it starts a new coroutine in the specified [context].
- * Coroutine emits ([Subscriber.onNext]) values with `send`, completes ([Subscriber.onComplete])
- * when the coroutine completes or channel is explicitly closed and emits error ([Subscriber.onError])
- * if coroutine throws an exception or closes channel with a cause.
- * Unsubscribing cancels running coroutine.
+ * The coroutine emits (via [Flow.Subscriber.onNext]) values with [send][ProducerScope.send],
+ * completes (via [Flow.Subscriber.onComplete]) when the coroutine completes or channel is explicitly closed, and emits
+ * errors (via [Flow.Subscriber.onError]) if the coroutine throws an exception or closes channel with a cause.
+ * Unsubscribing cancels the running coroutine.
  *
- * Invocations of `send` are suspended appropriately when subscribers apply back-pressure and to ensure that
- * `onNext` is not invoked concurrently.
+ * Invocations of [send][ProducerScope.send] are suspended appropriately when subscribers apply back-pressure and to
+ * ensure that [onNext][Flow.Subscriber.onNext] is not invoked concurrently.
  *
  * Coroutine context can be specified with [context] argument.
- * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [Dispatchers.Default] is used.
- * Method throws [IllegalArgumentException] if provided [context] contains a [Job] instance.
+ * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [Dispatchers.Default] is
+ * used.
  *
  * **Note: This is an experimental api.** Behaviour of publishers that work as children in a parent scope with respect
  *        to cancellation and error handling may change in the future.
+ *
+ * @throws IllegalArgumentException if the provided [context] contains a [Job] instance.
  */
-@ExperimentalCoroutinesApi // Since 1.3.x
+@ExperimentalCoroutinesApi
 public fun <T> flowPublish(
     context: CoroutineContext = EmptyCoroutineContext,
     @BuilderInference block: suspend ProducerScope<T>.() -> Unit
-): Flow.Publisher<T> {
-    val reactivePublisher : org.reactivestreams.Publisher<T> = kotlinx.coroutines.reactive.publish<T>(context, block)
-    return FlowAdapters.toFlowPublisher(reactivePublisher)
-}
+): Flow.Publisher<T> = FlowAdapters.toFlowPublisher(publish(context, block))

reactive/kotlinx-coroutines-jdk9/src/ReactiveFlow.kt

@@ -7,41 +7,43 @@ package kotlinx.coroutines.jdk9
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.reactive.asFlow
-import kotlinx.coroutines.reactive.asPublisher
+import kotlinx.coroutines.reactive.asPublisher as asReactivePublisher
 import kotlinx.coroutines.reactive.collect
+import kotlinx.coroutines.channels.*
 import org.reactivestreams.*
 import kotlin.coroutines.*
 import java.util.concurrent.Flow as JFlow
 
 /**
- * Transforms the given reactive [Publisher] into [Flow].
- * Use [buffer] operator on the resulting flow to specify the size of the backpressure.
- * More precisely, it specifies the value of the subscription's [request][Subscription.request].
- * [buffer] default capacity is used by default.
+ * Transforms the given reactive [Flow Publisher][JFlow.Publisher] into [Flow].
+ * Use the [buffer] operator on the resulting flow to specify the size of the back-pressure.
+ * More precisely, it specifies the value of the subscription's [request][JFlow.Subscription.request].
+ * The [default buffer capacity][Channel.BUFFERED] for a suspending channel is used by default.
  *
- * If any of the resulting flow transformations fails, subscription is immediately cancelled and all in-flight elements
- * are discarded.
+ * If any of the resulting flow transformations fails, the subscription is immediately cancelled and all the in-flight
+ * elements are discarded.
  */
 public fun <T : Any> JFlow.Publisher<T>.asFlow(): Flow<T> =
-        FlowAdapters.toPublisher(this).asFlow()
+    FlowAdapters.toPublisher(this).asFlow()
 
 /**
- * Transforms the given flow to a reactive specification compliant [Publisher].
+ * Transforms the given flow into a reactive specification compliant [Flow Publisher][JFlow.Publisher].
  *
- * An optional [context] can be specified to control the execution context of calls to [Subscriber] methods.
- * You can set a [CoroutineDispatcher] to confine them to a specific thread and/or various [ThreadContextElement] to
+ * An optional [context] can be specified to control the execution context of calls to the [Flow Subscriber][Subscriber]
+ * methods.
+ * A [CoroutineDispatcher] can be set to confine them to a specific thread; various [ThreadContextElement] can be set to
  * inject additional context into the caller thread. By default, the [Unconfined][Dispatchers.Unconfined] dispatcher
  * is used, so calls are performed from an arbitrary thread.
  */
 @JvmOverloads // binary compatibility
-public fun <T : Any> Flow<T>.asPublisher(context: CoroutineContext = EmptyCoroutineContext): JFlow.Publisher<T> {
-    val reactivePublisher : org.reactivestreams.Publisher<T> = this.asPublisher<T>(context)
-    return FlowAdapters.toFlowPublisher(reactivePublisher)
-}
+public fun <T : Any> Flow<T>.asPublisher(context: CoroutineContext = EmptyCoroutineContext): JFlow.Publisher<T> =
+    FlowAdapters.toFlowPublisher(asReactivePublisher(context))
 
 /**
- * Subscribes to this [Publisher] and performs the specified action for each received element.
- * Cancels subscription if any exception happens during collect.
+ * Subscribes to this [Flow Publisher][JFlow.Publisher] and performs the specified action for each received element.
+ *
+ * If [action] throws an exception at some point, the subscription is cancelled, and the exception is rethrown from
+ * [collect]. Also, if the publisher signals an error, that error is rethrown from [collect].
  */
 public suspend inline fun <T> JFlow.Publisher<T>.collect(action: (T) -> Unit): Unit =
     FlowAdapters.toPublisher(this).collect(action)