-
Notifications
You must be signed in to change notification settings - Fork 1.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Reactor coroutine context propagation in more places
* Propagation of the coroutine context of await calls into Mono/Flux builder * Publisher.asFlow propagates coroutine context from `collect` call to the Publisher * Flow.asFlux transform Fixes #284
- Loading branch information
1 parent
1156e1c
commit 4c069fc
Showing
22 changed files
with
347 additions
and
134 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
14 changes: 14 additions & 0 deletions
14
reactive/kotlinx-coroutines-reactive/src/ContextInjector.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,14 @@ | ||
| package kotlinx.coroutines.reactive | ||
|
|
||
| import kotlinx.coroutines.InternalCoroutinesApi | ||
| import org.reactivestreams.Publisher | ||
| import kotlin.coroutines.CoroutineContext | ||
|
|
||
| /** @suppress */ | ||
| @InternalCoroutinesApi | ||
| public interface ContextInjector { | ||
| /** | ||
| * Injects the coroutine context into the context of the publisher. | ||
| */ | ||
| public fun <T> injectCoroutineContext(publisher: Publisher<T>, coroutineContext: CoroutineContext): Publisher<T> | ||
| } |
109 changes: 109 additions & 0 deletions
109
reactive/kotlinx-coroutines-reactive/src/FlowAsPublisher.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,109 @@ | ||
| /* | ||
| * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. | ||
| */ | ||
|
|
||
| @file:JvmMultifileClass | ||
| @file:JvmName("FlowKt") | ||
|
|
||
| package kotlinx.coroutines.reactive | ||
|
|
||
| import kotlinx.coroutines.* | ||
| import kotlinx.coroutines.flow.* | ||
| import org.reactivestreams.* | ||
| import java.util.concurrent.atomic.* | ||
| import kotlin.coroutines.* | ||
|
|
||
| /** | ||
| * Transforms the given flow to a spec-compliant [Publisher]. | ||
| */ | ||
| @ExperimentalCoroutinesApi | ||
| public fun <T : Any> Flow<T>.asPublisher(): Publisher<T> = FlowAsPublisher(this) | ||
|
|
||
| /** | ||
| * Adapter that transforms [Flow] into TCK-complaint [Publisher]. | ||
| * [cancel] invocation cancels the original flow. | ||
| */ | ||
| @Suppress("PublisherImplementation") | ||
| private class FlowAsPublisher<T : Any>(private val flow: Flow<T>) : Publisher<T> { | ||
| override fun subscribe(subscriber: Subscriber<in T>?) { | ||
| if (subscriber == null) throw NullPointerException() | ||
| subscriber.onSubscribe(FlowSubscription(flow, subscriber)) | ||
| } | ||
| } | ||
|
|
||
| /** @suppress */ | ||
| @InternalCoroutinesApi | ||
| public class FlowSubscription<T>( | ||
| @JvmField val flow: Flow<T>, | ||
| @JvmField val subscriber: Subscriber<in T> | ||
| ) : Subscription { | ||
| @Volatile | ||
| private var canceled: Boolean = false | ||
| private val requested = AtomicLong(0L) | ||
| private val producer: AtomicReference<CancellableContinuation<Unit>?> = AtomicReference() | ||
|
|
||
| // This is actually optimizable | ||
| private var job = GlobalScope.launch(Dispatchers.Unconfined, start = CoroutineStart.LAZY) { | ||
| try { | ||
| consumeFlow() | ||
| subscriber.onComplete() | ||
| } catch (e: Throwable) { | ||
| // Failed with real exception, not due to cancellation | ||
| if (!coroutineContext[Job]!!.isCancelled) { | ||
| subscriber.onError(e) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private suspend fun consumeFlow() { | ||
| flow.collect { value -> | ||
| if (!coroutineContext.isActive) { | ||
| subscriber.onComplete() | ||
| coroutineContext.ensureActive() | ||
| } | ||
|
|
||
| if (requested.get() == 0L) { | ||
| suspendCancellableCoroutine<Unit> { | ||
| producer.set(it) | ||
| if (requested.get() != 0L) it.resumeSafely() | ||
| } | ||
| } | ||
|
|
||
| requested.decrementAndGet() | ||
| subscriber.onNext(value) | ||
| } | ||
| } | ||
|
|
||
| override fun cancel() { | ||
| canceled = true | ||
| job.cancel() | ||
| } | ||
|
|
||
| override fun request(n: Long) { | ||
| if (n <= 0) { | ||
| return | ||
| } | ||
|
|
||
| if (canceled) return | ||
|
|
||
| job.start() | ||
| var snapshot: Long | ||
| var newValue: Long | ||
| do { | ||
| snapshot = requested.get() | ||
| newValue = snapshot + n | ||
| if (newValue <= 0L) newValue = Long.MAX_VALUE | ||
| } while (!requested.compareAndSet(snapshot, newValue)) | ||
|
|
||
| val prev = producer.get() | ||
| if (prev == null || !producer.compareAndSet(prev, null)) return | ||
| prev.resumeSafely() | ||
| } | ||
|
|
||
| private fun CancellableContinuation<Unit>.resumeSafely() { | ||
| val token = tryResume(Unit) | ||
| if (token != null) { | ||
| completeResume(token) | ||
| } | ||
| } | ||
| } |
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
103 changes: 0 additions & 103 deletions
103
reactive/kotlinx-coroutines-reactive/src/flow/FlowAsPublisher.kt
This file was deleted.
Oops, something went wrong.
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
Oops, something went wrong.