Initial fix for #10.

Author: konrad-kaminski

build.gradle

@@ -285,6 +285,7 @@ project(":spring-webflux-kotlin-coroutine") {
         testCompile "org.springframework.boot:spring-boot-autoconfigure:$spring_boot_version"
         testCompile "org.springframework.boot:spring-boot-starter-test:$spring_boot_version"
         testCompile "org.springframework.boot:spring-boot-starter-webflux:$spring_boot_version"
+        testCompile "org.springframework.boot:spring-boot-starter-mustache:$spring_boot_version"
         testCompile "de.flapdoodle.embed:de.flapdoodle.embed.mongo:2.0.0"
     }
 

spring-kotlin-coroutine/src/main/kotlin/org/springframework/kotlin/experimental/coroutine/EnableCoroutine.kt

@@ -24,12 +24,13 @@ import org.springframework.kotlin.experimental.coroutine.context.CoroutineConfig
 import org.springframework.kotlin.experimental.coroutine.context.CoroutineContextResolverConfiguration
 import org.springframework.kotlin.experimental.coroutine.context.CoroutineContexts
 import org.springframework.kotlin.experimental.coroutine.event.CoroutineEventSupportConfiguraton
+import org.springframework.kotlin.experimental.coroutine.reactive.CoroutineReactiveAdapterRegistryConfiguration
 import org.springframework.kotlin.experimental.coroutine.scheduler.CoroutineSchedulerConfiguration
 
 @Target(AnnotationTarget.CLASS)
 @Retention(AnnotationRetention.RUNTIME)
 @MustBeDocumented
-@Import(CoroutineConfigurationSelector::class,
+@Import(CoroutineConfigurationSelector::class, CoroutineReactiveAdapterRegistryConfiguration::class,
         CoroutineContextResolverConfiguration::class, CoroutineContexts::class,
         CoroutineEventSupportConfiguraton::class, CoroutineCacheConfiguration::class,
         CoroutineSchedulerConfiguration::class)

spring-kotlin-coroutine/src/main/kotlin/org/springframework/kotlin/experimental/coroutine/SpringUtil.kt

@@ -1,8 +1,14 @@
 package org.springframework.kotlin.experimental.coroutine
 
+import kotlinx.coroutines.experimental.channels.Channel
+import kotlinx.coroutines.experimental.channels.ReceiveChannel
+import kotlinx.coroutines.experimental.reactive.asPublisher
+import kotlinx.coroutines.experimental.reactive.openSubscription
 import org.springframework.context.annotation.Condition
 import org.springframework.context.annotation.ConditionContext
 import org.springframework.context.annotation.Conditional
+import org.springframework.core.ReactiveAdapterRegistry
+import org.springframework.core.ReactiveTypeDescriptor
 import org.springframework.core.type.AnnotatedTypeMetadata
 import java.lang.annotation.Inherited
 
@@ -26,4 +32,11 @@ internal open class OnClassCondition: Condition {
             } catch (e: ClassNotFoundException) {
                 false
             }
+}
+
+fun ReactiveAdapterRegistry.registerReceiveChannel() {
+    registerReactiveType(
+            ReactiveTypeDescriptor.multiValue(ReceiveChannel::class.java) {
+                Channel<Any>().apply { close() }
+            }, { (it as ReceiveChannel<*>).asPublisher() }, { it.openSubscription() })
 }

spring-kotlin-coroutine/src/main/kotlin/org/springframework/kotlin/experimental/coroutine/reactive/CoroutineReactiveAdapterRegistryConfiguration.kt

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kotlin.experimental.coroutine.reactive
+
+import org.springframework.beans.factory.InitializingBean
+import org.springframework.beans.factory.config.BeanDefinition
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.context.annotation.Role
+import org.springframework.core.ReactiveAdapterRegistry
+import org.springframework.kotlin.experimental.coroutine.registerReceiveChannel
+
+@Configuration
+@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+internal open class CoroutineReactiveAdapterRegistryConfiguration {
+    @Bean
+    open fun reactiveAdapterRegistryInit(): InitializingBean = InitializingBean {
+        ReactiveAdapterRegistry.getSharedInstance().registerReceiveChannel()
+    }
+}

spring-webflux-kotlin-coroutine/src/main/kotlin/org/springframework/kotlin/experimental/coroutine/web/CoroutinesWebFluxConfigurer.kt

@@ -35,6 +35,7 @@ import org.springframework.web.reactive.result.method.annotation.injectCustomCon
 import org.springframework.web.server.ServerWebExchange
 import reactor.core.publisher.Mono
 import java.util.concurrent.CompletableFuture
+import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.experimental.Continuation
 import kotlin.coroutines.experimental.CoroutineContext
 import kotlin.coroutines.experimental.EmptyCoroutineContext
@@ -57,7 +58,7 @@ open class CoroutinesWebFluxConfigurer(
             override fun resolveArgument(parameter: MethodParameter, bindingContext: BindingContext,
                                          exchange: ServerWebExchange): Mono<Any> =
                 CompletableFutureContinuation().let {
-                    bindingContext.model.addAttribute("continuation", it)
+                    bindingContext.model.addAttribute("__continuation", AtomicReference(it))
 
                     Mono.just(it)
                 }

spring-webflux-kotlin-coroutine/src/main/kotlin/org/springframework/web/reactive/result/method/annotation/CustomControllerMethodResolver.kt

@@ -22,7 +22,6 @@ import kotlinx.coroutines.experimental.reactive.asPublisher
 import org.reactivestreams.Publisher
 import org.springframework.context.ConfigurableApplicationContext
 import org.springframework.core.ReactiveAdapterRegistry
-import org.springframework.kotlin.experimental.coroutine.isCoroutineCollection
 import org.springframework.kotlin.experimental.coroutine.returnTypeMetadata
 import org.springframework.util.ClassUtils
 import org.springframework.web.method.HandlerMethod
@@ -31,12 +30,11 @@ import org.springframework.web.reactive.HandlerResult
 import org.springframework.web.reactive.result.method.InvocableHandlerMethod
 import org.springframework.web.reactive.result.method.SyncInvocableHandlerMethod
 import org.springframework.web.server.ServerWebExchange
-import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import java.lang.reflect.Proxy
 import java.util.concurrent.CompletableFuture
+import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED
-import kotlin.reflect.jvm.kotlinFunction
 
 private class CustomControllerMethodResolver(
         private val delegate: ControllerMethodResolver
@@ -78,7 +76,7 @@ private class CoroutineInvocableHandlerMethod(private val handlerMethod: Handler
             super.invoke(exchange, bindingContext, *providedArgs)
                     .map { result ->
                         if (result.returnValue === COROUTINE_SUSPENDED) {
-                            val future = bindingContext.model.asMap()["continuation"] as CompletableFuture<*>
+                            val future = (bindingContext.model.asMap()["__continuation"] as AtomicReference<CompletableFuture<*>>).get()
                             val metaData = handlerMethod.method.returnTypeMetadata
                             val value = Mono.fromFuture(future).let { mono ->
                                     if (ClassUtils.isAssignable(ReceiveChannel::class.java, metaData.clazz)) {

spring-webflux-kotlin-coroutine/src/test/groovy/org/springframework/kotlin/experimental/coroutine/web/WebConfiguration.groovy

@@ -23,4 +23,9 @@ class WebConfiguration {
     CoroutineController coroutineController() {
         return new CoroutineController()
     }
+
+    @Bean
+    HtmlController htmlController() {
+        return new HtmlController()
+    }
 }

spring-webflux-kotlin-coroutine/src/test/groovy/org/springframework/kotlin/experimental/coroutine/web/WebIntSpec.groovy

@@ -25,6 +25,7 @@ import org.springframework.http.HttpStatus
 import org.springframework.kotlin.experimental.coroutine.IntSpecConfiguration
 import org.springframework.web.client.RestOperations
 import org.springframework.web.client.RestTemplate
+import spock.lang.Ignore
 import spock.lang.Specification
 import spock.lang.Unroll
 
@@ -90,4 +91,23 @@ data:4"""
         result.statusCode == HttpStatus.OK
         result.body == "123456"
     }
+
+    def "should handle suspending controller functions which return view name"() {
+        when:
+        def result = restTemplate.getForEntity("http://localhost:$port/blog", String)
+
+        then:
+        result.statusCode == HttpStatus.OK
+        result.body.trim() == "<article><title>TestTitle</title><text>TestText</text></article>"
+    }
+
+    @Ignore("Waiting for the fix in Spring 5")
+    def "should handle suspending controller functions which return view name different from endpoint path"() {
+        when:
+        def result = restTemplate.getForEntity("http://localhost:$port/blogEndpoint", String)
+
+        then:
+        result.statusCode == HttpStatus.OK
+        result.body.trim() == "<article><title>TestTitle</title><text>TestText</text></article>"
+    }
 }

spring-webflux-kotlin-coroutine/src/test/kotlin/org/springframework/kotlin/experimental/coroutine/web/HtmlController.kt

@@ -0,0 +1,28 @@
+package org.springframework.kotlin.experimental.coroutine.web
+
+import kotlinx.coroutines.experimental.channels.Channel
+import org.springframework.stereotype.Controller
+import org.springframework.ui.Model
+import org.springframework.ui.set
+import org.springframework.web.bind.annotation.GetMapping
+
+@Controller
+open class HtmlController {
+    @GetMapping("/blog")
+    suspend fun blog(model: Model): String {
+        model["articles"] = Channel<Article>(1).apply {
+            send(Article("TestTitle", "TestText"))
+            close()
+        }
+
+        return "blog"
+    }
+
+    @GetMapping("/blogEndpoint")
+    suspend fun blogEndpoint(model: Model): String = blog(model)
+}
+
+data class Article(
+    val title: String,
+    val text: String
+)

spring-webflux-kotlin-coroutine/src/test/resources/templates/blog.mustache

@@ -0,0 +1,3 @@
+{{#articles}}
+<article><title>{{title}}</title><text>{{text}}</text></article>
+{{/articles}}