|
| 1 | +/* |
| 2 | + * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. |
| 3 | + */ |
| 4 | + |
| 5 | +@file:JvmMultifileClass |
| 6 | +@file:JvmName("FlowKt") |
| 7 | + |
| 8 | +package kotlinx.coroutines.reactive |
| 9 | + |
| 10 | +import kotlinx.atomicfu.* |
| 11 | +import kotlinx.coroutines.* |
| 12 | +import kotlinx.coroutines.flow.* |
| 13 | +import org.reactivestreams.* |
| 14 | +import kotlinx.coroutines.intrinsics.* |
| 15 | + |
| 16 | +/** |
| 17 | + * Transforms the given flow to a spec-compliant [Publisher]. |
| 18 | + */ |
| 19 | +@ExperimentalCoroutinesApi |
| 20 | +public fun <T : Any> Flow<T>.asPublisher(): Publisher<T> = FlowAsPublisher(this) |
| 21 | + |
| 22 | +/** |
| 23 | + * Adapter that transforms [Flow] into TCK-complaint [Publisher]. |
| 24 | + * [cancel] invocation cancels the original flow. |
| 25 | + */ |
| 26 | +@Suppress("PublisherImplementation") |
| 27 | +private class FlowAsPublisher<T : Any>(private val flow: Flow<T>) : Publisher<T> { |
| 28 | + override fun subscribe(subscriber: Subscriber<in T>?) { |
| 29 | + if (subscriber == null) throw NullPointerException() |
| 30 | + subscriber.onSubscribe(FlowSubscription(flow, subscriber)) |
| 31 | + } |
| 32 | +} |
| 33 | + |
| 34 | +/** @suppress */ |
| 35 | +@InternalCoroutinesApi |
| 36 | +public class FlowSubscription<T>( |
| 37 | + @JvmField val flow: Flow<T>, |
| 38 | + @JvmField val subscriber: Subscriber<in T> |
| 39 | +) : Subscription, AbstractCoroutine<Unit>(Dispatchers.Unconfined, false) { |
| 40 | + private val requested = atomic(0L) |
| 41 | + private val producer = atomic<CancellableContinuation<Unit>?>(null) |
| 42 | + |
| 43 | + override fun onStart() { |
| 44 | + ::flowProcessing.startCoroutineCancellable(this) |
| 45 | + } |
| 46 | + |
| 47 | + private suspend fun flowProcessing() { |
| 48 | + try { |
| 49 | + consumeFlow() |
| 50 | + subscriber.onComplete() |
| 51 | + } catch (e: Throwable) { |
| 52 | + try { |
| 53 | + if (e is CancellationException) { |
| 54 | + subscriber.onComplete() |
| 55 | + } else { |
| 56 | + subscriber.onError(e) |
| 57 | + } |
| 58 | + } catch (e: Throwable) { |
| 59 | + // Last ditch report |
| 60 | + handleCoroutineException(coroutineContext, e) |
| 61 | + } |
| 62 | + } |
| 63 | + } |
| 64 | + |
| 65 | + /* |
| 66 | + * This method has at most one caller at any time (triggered from the `request` method) |
| 67 | + */ |
| 68 | + private suspend fun consumeFlow() { |
| 69 | + flow.collect { value -> |
| 70 | + /* |
| 71 | + * Flow is scopeless, thus if it's not active, its subscription was cancelled. |
| 72 | + * No intermediate "child failed, but flow coroutine is not" states are allowed. |
| 73 | + */ |
| 74 | + coroutineContext.ensureActive() |
| 75 | + if (requested.value <= 0L) { |
| 76 | + suspendCancellableCoroutine<Unit> { |
| 77 | + producer.value = it |
| 78 | + if (requested.value != 0L) it.resumeSafely() |
| 79 | + } |
| 80 | + } |
| 81 | + requested.decrementAndGet() |
| 82 | + subscriber.onNext(value) |
| 83 | + } |
| 84 | + } |
| 85 | + |
| 86 | + override fun cancel() { |
| 87 | + cancel(null) |
| 88 | + } |
| 89 | + |
| 90 | + override fun request(n: Long) { |
| 91 | + if (n <= 0) { |
| 92 | + return |
| 93 | + } |
| 94 | + start() |
| 95 | + requested.update { value -> |
| 96 | + val newValue = value + n |
| 97 | + if (newValue <= 0L) Long.MAX_VALUE else newValue |
| 98 | + } |
| 99 | + val producer = producer.getAndSet(null) ?: return |
| 100 | + producer.resumeSafely() |
| 101 | + } |
| 102 | + |
| 103 | + private fun CancellableContinuation<Unit>.resumeSafely() { |
| 104 | + val token = tryResume(Unit) |
| 105 | + if (token != null) { |
| 106 | + completeResume(token) |
| 107 | + } |
| 108 | + } |
| 109 | +} |
0 commit comments