Skip to content

Commit 1382220

Browse files
committed
Add support for Deferred to ReactiveAdapterRegistry
See gh-19975
1 parent 1c9cbaf commit 1382220

File tree

5 files changed

+107
-3
lines changed

5 files changed

+107
-3
lines changed

spring-core/spring-core.gradle

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,8 @@ dependencies {
7878
optional("io.reactivex:rxjava-reactive-streams:${rxjavaAdapterVersion}")
7979
optional("io.reactivex.rxjava2:rxjava:${rxjava2Version}")
8080
optional("io.netty:netty-buffer")
81+
optional("org.jetbrains.kotlinx:kotlinx-coroutines-core:${coroutinesVersion}")
82+
optional("org.jetbrains.kotlinx:kotlinx-coroutines-reactor:${coroutinesVersion}")
8183
testCompile("io.projectreactor:reactor-test")
8284
testCompile("org.apache.tomcat.embed:tomcat-embed-core:${tomcatVersion}")
8385
testCompile("org.xmlunit:xmlunit-matchers:2.6.2")

spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,8 +40,8 @@
4040
* {@code Observable}, and others.
4141
*
4242
* <p>By default, depending on classpath availability, adapters are registered
43-
* for Reactor, RxJava 1, RxJava 2 types, {@link CompletableFuture}, and Java 9+
44-
* Flow.Publisher.
43+
* for Reactor, RxJava 1, RxJava 2 types, {@link CompletableFuture}, Java 9+
44+
* {@code Flow.Publisher} and Kotlin Coroutines {@code Deferred}.
4545
*
4646
* @author Rossen Stoyanchev
4747
* @author Sebastien Deleuze
@@ -90,6 +90,11 @@ public ReactiveAdapterRegistry() {
9090
}
9191
// If not present, do nothing for the time being...
9292
// We can fall back on "reactive-streams-flow-bridge" (once released)
93+
94+
// Coroutines
95+
if (ClassUtils.isPresent("kotlinx.coroutines.Deferred", classLoader)) {
96+
CoroutinesRegistrarKt.registerAdapter(this);
97+
}
9398
}
9499

95100

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
/*
2+
* Copyright 2002-2019 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.core
18+
19+
import kotlinx.coroutines.Deferred
20+
import kotlinx.coroutines.GlobalScope
21+
import kotlinx.coroutines.async
22+
import kotlinx.coroutines.reactive.awaitFirstOrNull
23+
import kotlinx.coroutines.reactor.mono
24+
import reactor.core.publisher.toMono
25+
26+
/**
27+
* Register Reactive adapters for Coroutines types.
28+
*
29+
* @author Sebastien Deleuze
30+
* @since 5.2
31+
*/
32+
internal fun registerAdapter(registry: ReactiveAdapterRegistry) {
33+
registry.registerReactiveType(
34+
ReactiveTypeDescriptor.singleOptionalValue(Deferred::class.java) { GlobalScope.async {} },
35+
{ source -> GlobalScope.mono { (source as Deferred<*>).await() }},
36+
{ source -> GlobalScope.async { source.toMono().awaitFirstOrNull() } }
37+
)
38+
}

spring-core/src/test/java/org/springframework/core/ReactiveAdapterRegistryTests.java

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2002-2018 the original author or authors.
2+
* Copyright 2002-2019 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@
2323

2424
import io.reactivex.Flowable;
2525
import io.reactivex.Maybe;
26+
import kotlinx.coroutines.Deferred;
2627
import org.junit.Test;
2728
import org.reactivestreams.Publisher;
2829
import reactor.core.publisher.Flux;
@@ -68,6 +69,9 @@ public void defaultAdapterRegistrations() {
6869
assertNotNull(getAdapter(io.reactivex.Single.class));
6970
assertNotNull(getAdapter(Maybe.class));
7071
assertNotNull(getAdapter(io.reactivex.Completable.class));
72+
73+
// Coroutines
74+
assertNotNull(getAdapter(Deferred.class));
7175
}
7276

7377
@Test
Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
/*
2+
* Copyright 2002-2019 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.core
18+
19+
import kotlinx.coroutines.Deferred
20+
import kotlinx.coroutines.GlobalScope
21+
import kotlinx.coroutines.async
22+
import kotlinx.coroutines.runBlocking
23+
import org.junit.Assert.assertEquals
24+
import org.junit.Assert.assertTrue
25+
import org.junit.Test
26+
import org.reactivestreams.Publisher
27+
import reactor.core.publisher.Mono
28+
import java.time.Duration
29+
import kotlin.reflect.KClass
30+
31+
class KotlinReactiveAdapterRegistryTests {
32+
33+
private val registry = ReactiveAdapterRegistry.getSharedInstance()
34+
35+
@Test
36+
fun deferredToPublisher() {
37+
val source = GlobalScope.async { 1 }
38+
val target: Publisher<Int> = getAdapter(Deferred::class).toPublisher(source)
39+
assertTrue("Expected Mono Publisher: " + target.javaClass.name, target is Mono<*>)
40+
assertEquals(1, (target as Mono<Int>).block(Duration.ofMillis(1000)))
41+
}
42+
43+
@Test
44+
fun publisherToDeferred() {
45+
val source = Mono.just(1)
46+
val target = getAdapter(Deferred::class).fromPublisher(source)
47+
assertTrue(target is Deferred<*>)
48+
assertEquals(1, runBlocking { (target as Deferred<*>).await() })
49+
50+
}
51+
52+
private fun getAdapter(reactiveType: KClass<*>): ReactiveAdapter {
53+
return this.registry.getAdapter(reactiveType.java)!!
54+
}
55+
}

0 commit comments

Comments
 (0)