Skip to content

Promote reactive bridges for Flow to stable API #1549

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Sep 19, 2019
Merged

Promote reactive bridges for Flow to stable API #1549

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
46c18e3
Promote reactive adapters for Flow to stable API
qwwdfsad Sep 13, 2019
5410890
Promote Publisher<T>.collect to stable API
qwwdfsad Sep 13, 2019
dd8860f
Promote rx2 extensions to stable, increase deprecation level for obso…
qwwdfsad Sep 19, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion reactive/kotlinx-coroutines-reactive/src/Channel.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@ public suspend inline fun <T> Publisher<T>.consumeEach(action: (T) -> Unit) =
* Subscribes to this [Publisher] and performs the specified action for each received element.
* Cancels subscription if any exception happens during collect.
*/
@ExperimentalCoroutinesApi // Since 1.2.1, tentatively till 1.3.0
public suspend inline fun <T> Publisher<T>.collect(action: (T) -> Unit) =
openSubscription().consumeEach(action)

Expand Down
2 changes: 1 addition & 1 deletion reactive/kotlinx-coroutines-reactive/src/Publish.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ public fun <T> publish(

@Deprecated(
message = "CoroutineScope.publish is deprecated in favour of top-level publish",
level = DeprecationLevel.WARNING,
level = DeprecationLevel.ERROR,
replaceWith = ReplaceWith("publish(context, block)")
) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0. Binary compatibility with Spring
@LowPriorityInOverloadResolution
Expand Down
2 changes: 0 additions & 2 deletions reactive/kotlinx-coroutines-reactive/src/ReactiveFlow.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,12 @@ import kotlin.coroutines.*
* If any of the resulting flow transformations fails, subscription is immediately cancelled and all in-flights elements
* are discarded.
*/
@ExperimentalCoroutinesApi
public fun <T : Any> Publisher<T>.asFlow(): Flow<T> =
PublisherAsFlow(this, 1)

/**
* Transforms the given flow to a spec-compliant [Publisher].
*/
@ExperimentalCoroutinesApi
public fun <T : Any> Flow<T>.asPublisher(): Publisher<T> = FlowAsPublisher(this)

private class PublisherAsFlow<T : Any>(
Expand Down
2 changes: 0 additions & 2 deletions reactive/kotlinx-coroutines-reactor/src/ReactorFlow.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@

package kotlinx.coroutines.reactor

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.reactive.FlowSubscription
Expand All @@ -15,7 +14,6 @@ import reactor.core.publisher.Flux
* Converts the given flow to a cold flux.
* The original flow is cancelled when the flux subscriber is disposed.
*/
@ExperimentalCoroutinesApi
public fun <T: Any> Flow<T>.asFlux(): Flux<T> = FlowAsFlux(this)

private class FlowAsFlux<T : Any>(private val flow: Flow<T>) : Flux<T>() {
Expand Down