Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Context passing between coroutines and Reactor Mono/Flux
Fixes #284
- Loading branch information
1 parent
f22604b
commit ff06d83
Showing
8 changed files
with
147 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| package kotlinx.coroutines.reactor | ||
|
|
||
| import kotlinx.coroutines.ExperimentalCoroutinesApi | ||
| import reactor.util.context.Context | ||
| import kotlin.coroutines.* | ||
|
|
||
| /** | ||
| * Marks coroutine context element that contains Reactor's [Context] elements in [context] for seamless integration | ||
| * between [CoroutineContext] and Reactor's [Context]. | ||
| * | ||
| * [Context.asCoroutineContext] is defined to add Reactor's [Context] elements as part of [CoroutineContext]. | ||
| * | ||
| * Reactor builders: [mono], [flux] can extract the reactor context from their coroutine context and | ||
| * pass it on. Modifications of reactor context can be retrieved by `coroutineContext[ReactorContext]`. | ||
| * | ||
| * Example usage: | ||
| * | ||
| * Passing reactor context from coroutine builder to reactor entity: | ||
| * | ||
| * ``` | ||
| * launch(Context.of("key", "value").asCoroutineContext()) { | ||
| * mono { | ||
| * assertEquals(coroutineContext[ReactorContext]!!.context.get("key"), "value") | ||
| * }.subscribe() | ||
| * } | ||
| * ``` | ||
| * | ||
| * Accessing modified reactor context enriched from downstream via coroutine context: | ||
| * | ||
| * ``` | ||
| * launch { | ||
| * mono { | ||
| * assertEquals(coroutineContext[ReactorContext]!!.context.get("key"), "value") | ||
| * }.subscriberContext(Context.of("key", "value")) | ||
| * .subscribe() | ||
| * } | ||
| * ``` | ||
| */ | ||
| @ExperimentalCoroutinesApi | ||
| public class ReactorContext(val context: Context) : AbstractCoroutineContextElement(ReactorContext) { | ||
| companion object Key : CoroutineContext.Key<ReactorContext> | ||
| } | ||
|
|
||
|
|
||
| /** | ||
| * Wraps the given [Context] into [ReactorContext], so it can be added to coroutine's context | ||
| * and later retrieved via `coroutineContext[ReactorContext]`. | ||
| */ | ||
| @ExperimentalCoroutinesApi | ||
| public fun Context.asCoroutineContext(): CoroutineContext = ReactorContext(this) |
32 changes: 32 additions & 0 deletions
32
reactive/kotlinx-coroutines-reactor/test/ReactorContextTest.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| package kotlinx.coroutines.reactor | ||
|
|
||
| import kotlinx.coroutines.* | ||
| import kotlinx.coroutines.reactive.* | ||
| import org.junit.Test | ||
| import reactor.util.context.Context | ||
| import kotlin.test.assertEquals | ||
|
|
||
| class ReactorContextTest { | ||
| @Test | ||
| fun testMonoHookedContext() = runBlocking { | ||
| val mono = mono(Context.of(1, "1", 7, "7").asCoroutineContext()) { | ||
| val ctx = coroutineContext[ReactorContext]?.context | ||
| buildString { | ||
| (1..7).forEach { append(ctx?.getOrDefault(it, "noValue")) } | ||
| } | ||
| } .subscriberContext(Context.of(2, "2", 3, "3", 4, "4", 5, "5")) | ||
| .subscriberContext { ctx -> ctx.put(6, "6") } | ||
| assertEquals(mono.awaitFirst(), "1234567") | ||
| } | ||
|
|
||
| @Test | ||
| fun testFluxContext() = runBlocking<Unit> { | ||
| val flux = flux(Context.of(1, "1", 7, "7").asCoroutineContext()) { | ||
| val ctx = coroutineContext[ReactorContext]!!.context | ||
| (1..7).forEach { send(ctx.getOrDefault(it, "noValue")) } | ||
| } .subscriberContext(Context.of(2, "2", 3, "3", 4, "4", 5, "5")) | ||
| .subscriberContext { ctx -> ctx.put(6, "6") } | ||
| var i = 0 | ||
| flux.subscribe { str -> i++; assertEquals(str, i.toString()) } | ||
| } | ||
| } |