@@ -3,48 +3,48 @@ package kotlinx.coroutines.reactor
33import kotlinx.coroutines.ExperimentalCoroutinesApi
44import reactor.util.context.Context
55import kotlin.coroutines.*
6+ import kotlinx.coroutines.reactive.*
67
78/* *
89 * Wraps Reactor's [Context] into [CoroutineContext] element for seamless integration Reactor and kotlinx.coroutines.
9- *
1010 * [Context.asCoroutineContext] is defined to add Reactor's [Context] elements as part of [CoroutineContext].
11+ * Coroutine context element that propagates information about Reactor's [Context] through coroutines.
1112 *
12- * Reactor builders [mono] and [flux] use this context element to enhance the resulting `subscriberContext`.
13+ * This context element is implicitly propagated through subscriber's context by all Reactive integrations, such as [mono], [flux],
14+ * [Publisher.asFlow][asFlow], [Flow.asPublisher][asPublisher] and [Flow.asFlux][asFlux].
15+ * Functions that subscribe to the reactive stream (e.g. [Publisher.awaitFirst][awaitFirst]) also propagate [ReactorContext] to the
16+ * subscriber's [Context].
17+ **
18+ * ### Examples of Reactive context integration.
1319 *
14- * ### Usages
15- * Passing reactor context from coroutine builder to reactor entity:
16- * ```
17- * launch(Context.of("key", "value").asCoroutineContext()) {
18- * mono {
19- * println(coroutineContext[ReactorContext]) // Prints { "key": "value" }
20- * }.subscribe()
21- * }
20+ * #### Propagating ReactorContext to Reactor's Context
2221 * ```
22+ * val flux = myDatabaseService.getUsers()
23+ * .subscriberContext() { ctx -> println(ctx); ctx }
24+ * flux.await() // Will print "null"
2325 *
24- * Accessing modified reactor context enriched from the downstream:
25- * ```
26- * launch {
27- * mono {
28- * println(coroutineContext[ReactorContext]) // Prints { "key": "value" }
29- * }.subscriberContext(Context.of("key", "value"))
30- * .subscribe()
26+ * // Now add ReactorContext
27+ * withContext(Context.of("answer", "42").asCoroutineContext()) {
28+ * flux.await() // Will print "Context{'key'='value'}"
3129 * }
3230 * ```
3331 *
34- * [CoroutineContext] of a suspendable function that awaits a value from [Mono] or [Flux] instance
35- * is propagated into [mono] and [flux] Reactor builders:
32+ * #### Propagating subscriber's Context to ReactorContext:
3633 * ```
37- * launch(Context.of("key", "value").asCoroutineContext()) {
38- * assertEquals(bar().awaitFirst(), "value")
39- * }
40- *
41- * fun bar(): Mono<String> = mono {
42- * coroutineContext[ReactorContext]!!.context.get("key")
34+ * val flow = flow {
35+ * println("Reactor context in Flow: " + coroutineContext[ReactorContext])
4336 * }
37+ * // No context
38+ * flow.asFlux()
39+ * .subscribe() // Will print 'Reactor context in Flow: null'
40+ * // Add subscriber's context
41+ * flow.asFlux()
42+ * .subscriberContext { ctx -> ctx.put("answer", 42) }
43+ * .subscribe() // Will print "Reactor context in Flow: Context{'answer'=42}"
4444 * ```
4545 */
4646@ExperimentalCoroutinesApi
47- public class ReactorContext (val context : Context ) : AbstractCoroutineContextElement(ReactorContext ) {
47+ public class ReactorContext (public val context : Context ) : AbstractCoroutineContextElement(ReactorContext ) {
4848 companion object Key : CoroutineContext.Key<ReactorContext>
4949}
5050
@@ -53,4 +53,4 @@ public class ReactorContext(val context: Context) : AbstractCoroutineContextElem
5353 * and later used via `coroutineContext[ReactorContext]`.
5454 */
5555@ExperimentalCoroutinesApi
56- public fun Context.asCoroutineContext (): ReactorContext = ReactorContext (this )
56+ public fun Context.asCoroutineContext (): ReactorContext = ReactorContext (this )
0 commit comments