diff --git a/spring-cloud-function-kotlin/pom.xml b/spring-cloud-function-kotlin/pom.xml index 067674cbd..4eb9a24b7 100644 --- a/spring-cloud-function-kotlin/pom.xml +++ b/spring-cloud-function-kotlin/pom.xml @@ -24,11 +24,34 @@ org.jetbrains.kotlin kotlin-stdlib-jdk8 + + org.jetbrains.kotlin + kotlin-reflect + + + org.jetbrains.kotlinx + kotlinx-coroutines-reactor + org.springframework.boot spring-boot-starter-test test + + org.springframework.cloud + spring-cloud-function-web + test + + + org.springframework + spring-webflux + test + + + org.springframework.boot + spring-boot-starter-web + test + org.springframework.boot spring-boot-configuration-processor diff --git a/spring-cloud-function-kotlin/src/main/java/org/springframework/cloud/function/context/config/KotlinLambdaToFunctionAutoConfiguration.java b/spring-cloud-function-kotlin/src/main/java/org/springframework/cloud/function/context/config/KotlinLambdaToFunctionAutoConfiguration.java index 5e9371007..5f9cd2b8e 100644 --- a/spring-cloud-function-kotlin/src/main/java/org/springframework/cloud/function/context/config/KotlinLambdaToFunctionAutoConfiguration.java +++ b/spring-cloud-function-kotlin/src/main/java/org/springframework/cloud/function/context/config/KotlinLambdaToFunctionAutoConfiguration.java @@ -22,7 +22,6 @@ import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; - import kotlin.jvm.functions.Function0; import kotlin.jvm.functions.Function1; import kotlin.jvm.functions.Function2; @@ -30,6 +29,7 @@ import kotlin.jvm.functions.Function3; import kotlin.jvm.functions.Function4; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import reactor.core.publisher.Flux; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -50,11 +50,13 @@ import org.springframework.context.annotation.Configuration; import org.springframework.core.ResolvableType; import org.springframework.util.ObjectUtils; + /** * Configuration class which defines the required infrastructure to bootstrap Kotlin * lambdas as invocable functions within the context of the framework. * * @author Oleg Zhurakousky + * @author Adrien Poupard * @since 2.0 */ @Configuration @@ -147,11 +149,17 @@ public class KotlinLambdaToFunctionAutoConfiguration { @Override public Object invoke(Object arg0) { + if (CoroutinesUtils.isValidSuspendingFunction(kotlinLambdaTarget, arg0)) { + return CoroutinesUtils.invokeSuspendingFunction(kotlinLambdaTarget, arg0); + } return ((Function1) this.kotlinLambdaTarget).invoke(arg0); } @Override public Object invoke() { + if (CoroutinesUtils.isValidSuspendingSupplier(kotlinLambdaTarget)) { + return CoroutinesUtils.invokeSuspendingSupplier(kotlinLambdaTarget); + } return ((Function0) this.kotlinLambdaTarget).invoke(); } @@ -168,19 +176,43 @@ public class KotlinLambdaToFunctionAutoConfiguration { @Override public FunctionRegistration getObject() throws Exception { String name = this.name.endsWith(FunctionRegistration.REGISTRATION_NAME_SUFFIX) - ? this.name.replace(FunctionRegistration.REGISTRATION_NAME_SUFFIX, "") - : this.name; + ? this.name.replace(FunctionRegistration.REGISTRATION_NAME_SUFFIX, "") + : this.name; Type functionType = FunctionContextUtils.findType(name, this.beanFactory); FunctionRegistration registration = new FunctionRegistration<>(this, name); Type[] types = ((ParameterizedType) functionType).getActualTypeArguments(); if (functionType.getTypeName().contains("Function0")) { functionType = ResolvableType.forClassWithGenerics(Supplier.class, ResolvableType.forType(types[0])) - .getType(); + .getType(); + } - else if (functionType.getTypeName().contains("Function1")) { + else if (isValidKotlinFunction(functionType, types)) { functionType = ResolvableType.forClassWithGenerics(Function.class, ResolvableType.forType(types[0]), - ResolvableType.forType(types[1])).getType(); + ResolvableType.forType(types[1])).getType(); + } + else if (isValidKotlinSuspendSupplier(functionType, types)) { + Type continuationReturnType = CoroutinesUtils.getSuspendingFunctionReturnType(types[0]); + functionType = ResolvableType.forClassWithGenerics( + Supplier.class, + ResolvableType.forClassWithGenerics(Flux.class, ResolvableType.forType(continuationReturnType)) + ).getType(); + } + else if (isValidKotlinSuspendFunction(functionType, types)) { + Type continuationArgType = CoroutinesUtils.getSuspendingFunctionArgType(types[0]); + Type continuationReturnType = CoroutinesUtils.getSuspendingFunctionReturnType(types[1]); + functionType = ResolvableType.forClassWithGenerics( + Function.class, + ResolvableType.forClassWithGenerics(Flux.class, ResolvableType.forType(continuationArgType)), + ResolvableType.forClassWithGenerics(Flux.class, ResolvableType.forType(continuationReturnType)) + ).getType(); + } + else if (isValidKotlinSuspendConsumer(functionType, types)) { + Type continuationArgType = CoroutinesUtils.getSuspendingFunctionArgType(types[0]); + functionType = ResolvableType.forClassWithGenerics( + Consumer.class, + ResolvableType.forClassWithGenerics(Flux.class, ResolvableType.forType(continuationArgType)) + ).getType(); } else { throw new UnsupportedOperationException("Multi argument Kotlin functions are not currently supported"); @@ -189,6 +221,22 @@ public class KotlinLambdaToFunctionAutoConfiguration { return registration; } + private boolean isValidKotlinFunction(Type functionType, Type[] type) { + return functionType.getTypeName().contains(Function1.class.getName()) && type.length == 2 && !CoroutinesUtils.isContinuationType(type[0]); + } + + private boolean isValidKotlinSuspendSupplier(Type functionType, Type[] type) { + return functionType.getTypeName().contains(Function1.class.getName()) && type.length == 2 && CoroutinesUtils.isContinuationFlowType(type[0]); + } + + private boolean isValidKotlinSuspendConsumer(Type functionType, Type[] type) { + return functionType.getTypeName().contains(Function2.class.getName()) && type.length == 3 && CoroutinesUtils.isFlowType(type[0]) && CoroutinesUtils.isContinuationUnitType(type[1]); + } + + private boolean isValidKotlinSuspendFunction(Type functionType, Type[] type) { + return functionType.getTypeName().contains(Function2.class.getName()) && type.length == 3 && CoroutinesUtils.isContinuationFlowType(type[1]); + } + @Override public Class getObjectType() { return FunctionRegistration.class; diff --git a/spring-cloud-function-kotlin/src/main/kotlin/org/springframework/cloud/function/context/config/CoroutinesUtils.kt b/spring-cloud-function-kotlin/src/main/kotlin/org/springframework/cloud/function/context/config/CoroutinesUtils.kt new file mode 100644 index 000000000..8727cfe7d --- /dev/null +++ b/spring-cloud-function-kotlin/src/main/kotlin/org/springframework/cloud/function/context/config/CoroutinesUtils.kt @@ -0,0 +1,120 @@ +/* + * Copyright 2012-2021 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 + * + * https://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. + */ + +@file:JvmName("CoroutinesUtils") +package org.springframework.cloud.function.context.config + +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.reactive.asFlow +import kotlinx.coroutines.reactor.asFlux +import kotlinx.coroutines.reactor.mono +import reactor.core.publisher.Flux +import java.lang.reflect.ParameterizedType +import java.lang.reflect.Type +import java.lang.reflect.WildcardType +import kotlin.coroutines.Continuation +import kotlin.coroutines.intrinsics.suspendCoroutineUninterceptedOrReturn + +fun isValidSuspendingFunction(kotlinLambdaTarget: Any, arg0: Any): Boolean { + return arg0 is Flux<*> && kotlinLambdaTarget is Function2<*, *, *> +} + +fun getSuspendingFunctionArgType(type: Type): Type { + return getFlowTypeArguments(type) +} + +fun getFlowTypeArguments(type: Type): Type { + if(!isFlowType(type)) { + return type + } + val parameterizedLowerType = type as ParameterizedType + if(parameterizedLowerType.actualTypeArguments.isEmpty()) { + return parameterizedLowerType + } + + val actualTypeArgument = parameterizedLowerType.actualTypeArguments[0] + return if(actualTypeArgument is WildcardType) { + val wildcardTypeLower = parameterizedLowerType.actualTypeArguments[0] as WildcardType + wildcardTypeLower.upperBounds[0] + } else { + actualTypeArgument + } +} + +fun isFlowType(type: Type): Boolean { + return type.typeName.startsWith(Flow::class.qualifiedName!!) +} + +fun getSuspendingFunctionReturnType(type: Type): Type { + val lower = getContinuationTypeArguments(type) + return getFlowTypeArguments(lower) +} + +fun isContinuationType(type: Type): Boolean { + return type.typeName.startsWith(Continuation::class.qualifiedName!!) +} + +fun isContinuationUnitType(type: Type): Boolean { + return isContinuationType(type) && type.typeName.contains(Unit::class.qualifiedName!!) +} + +fun isContinuationFlowType(type: Type): Boolean { + return isContinuationType(type) && type.typeName.contains(Flow::class.qualifiedName!!) +} + +private fun getContinuationTypeArguments(type: Type): Type { + if(!isContinuationType(type)) { + return type + } + val parameterizedType = type as ParameterizedType + val wildcardType = parameterizedType.actualTypeArguments[0] as WildcardType + return wildcardType.lowerBounds[0] +} + +fun invokeSuspendingFunction(kotlinLambdaTarget: Any, arg0: Any): Flux { + val function = kotlinLambdaTarget as SuspendFunction + val flux = arg0 as Flux + return fluxSuspendingFlowFunction(flux, function) +} + +fun isValidSuspendingSupplier(kotlinLambdaTarget: Any): Boolean { + return kotlinLambdaTarget is Function1<*, *> +} + +fun invokeSuspendingSupplier(kotlinLambdaTarget: Any): Flux { + val supplier = kotlinLambdaTarget as SuspendSupplier + return mono(Dispatchers.Unconfined) { + suspendCoroutineUninterceptedOrReturn> { + supplier.invoke(it) + } + }.flatMapMany { + it.asFlux() + } +} + +fun fluxSuspendingFlowFunction(flux: Flux, target: SuspendFunction): Flux { + return mono(Dispatchers.Unconfined) { + suspendCoroutineUninterceptedOrReturn> { + target.invoke(flux.asFlow(), it) + } + }.flatMapMany { + it.asFlux() + } +} + +private typealias SuspendFunction = (Any?, Any?) -> Any? +private typealias SuspendSupplier = (Any?) -> Any? diff --git a/spring-cloud-function-kotlin/src/test/java/org/springframework/cloud/function/kotlin/ContextFunctionCatalogAutoConfigurationKotlinSuspendTests.java b/spring-cloud-function-kotlin/src/test/java/org/springframework/cloud/function/kotlin/ContextFunctionCatalogAutoConfigurationKotlinSuspendTests.java new file mode 100644 index 000000000..53450d2ba --- /dev/null +++ b/spring-cloud-function-kotlin/src/test/java/org/springframework/cloud/function/kotlin/ContextFunctionCatalogAutoConfigurationKotlinSuspendTests.java @@ -0,0 +1,126 @@ +/* + * Copyright 2019-2021 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 + * + * https://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.cloud.function.kotlin; + +import java.lang.reflect.ParameterizedType; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; + +import kotlin.jvm.functions.Function2; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.function.context.catalog.FunctionTypeUtils; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.support.GenericApplicationContext; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * @author Adrien Poupard + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class ContextFunctionCatalogAutoConfigurationKotlinSuspendTests { + + private GenericApplicationContext context; + + private FunctionCatalog catalog; + + @AfterEach + public void close() { + if (this.context != null) { + this.context.close(); + } + } + + @Test + public void typeDiscoveryTests() { + create(new Class[] { KotlinSuspendFlowLambdasConfiguration.class, + ContextFunctionCatalogAutoConfigurationKotlinTests.SimpleConfiguration.class }); + + Object function = this.context.getBean("kotlinFunction"); + ParameterizedType functionType = (ParameterizedType) FunctionTypeUtils.discoverFunctionType(function, "kotlinFunction", this.context); + assertThat(functionType.getRawType().getTypeName()).isEqualTo(Function.class.getName()); + assertThat(functionType.getActualTypeArguments().length).isEqualTo(2); + assertThat(functionType.getActualTypeArguments()[0].getTypeName()).isEqualTo("reactor.core.publisher.Flux"); + assertThat(functionType.getActualTypeArguments()[1].getTypeName()).isEqualTo("reactor.core.publisher.Flux"); + + function = this.context.getBean("kotlinConsumer"); + functionType = (ParameterizedType) FunctionTypeUtils.discoverFunctionType(function, "kotlinConsumer", this.context); + assertThat(functionType.getRawType().getTypeName()).isEqualTo(Consumer.class.getName()); + assertThat(functionType.getActualTypeArguments().length).isEqualTo(1); + assertThat(functionType.getActualTypeArguments()[0].getTypeName()).isEqualTo("reactor.core.publisher.Flux"); + + function = this.context.getBean("kotlinSupplier"); + functionType = (ParameterizedType) FunctionTypeUtils.discoverFunctionType(function, "kotlinSupplier", this.context); + assertThat(functionType.getRawType().getTypeName()).isEqualTo(Supplier.class.getName()); + assertThat(functionType.getActualTypeArguments().length).isEqualTo(1); + assertThat(functionType.getActualTypeArguments()[0].getTypeName()).isEqualTo("reactor.core.publisher.Flux"); + + function = this.context.getBean("kotlinPojoFunction"); + functionType = (ParameterizedType) FunctionTypeUtils.discoverFunctionType(function, "kotlinPojoFunction", this.context); + assertThat(functionType.getRawType().getTypeName()).isEqualTo(Function.class.getName()); + assertThat(functionType.getActualTypeArguments().length).isEqualTo(2); + assertThat(functionType.getActualTypeArguments()[0].getTypeName()).isEqualTo("reactor.core.publisher.Flux"); + assertThat(functionType.getActualTypeArguments()[1].getTypeName()).isEqualTo("reactor.core.publisher.Flux"); + } + + @Test + public void shouldNotLoadKotlinSuspendLambasNotUsingFlow() { + create(new Class[] { KotlinSuspendLambdasConfiguration.class, + ContextFunctionCatalogAutoConfigurationKotlinTests.SimpleConfiguration.class }); + + assertThat(this.context.getBean("kotlinFunction")).isInstanceOf(Function2.class); + assertThatThrownBy(() -> { + this.catalog.lookup(Function.class, "kotlinFunction"); + }).isInstanceOf(BeanCreationException.class); + + assertThatThrownBy(() -> { + this.catalog.lookup(Function.class, "kotlinConsumer"); + }).isInstanceOf(BeanCreationException.class); + + assertThatThrownBy(() -> { + this.catalog.lookup(Supplier.class, "kotlinSupplier"); + }).isInstanceOf(BeanCreationException.class); + + } + + private void create(Class[] types, String... props) { + this.context = (GenericApplicationContext) new SpringApplicationBuilder(types).properties(props).run(); + this.catalog = this.context.getBean(FunctionCatalog.class); + } + + @EnableAutoConfiguration + @Configuration + protected static class SimpleConfiguration { + + @Bean + public Function function2() { + return value -> value + "function2"; + } + + } + +} diff --git a/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/KotlinLambdasConfiguration.kt b/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/KotlinLambdasConfiguration.kt index 6c9063f3c..caaf8db0e 100644 --- a/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/KotlinLambdasConfiguration.kt +++ b/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/KotlinLambdasConfiguration.kt @@ -32,7 +32,7 @@ class KotlinLambdasConfiguration { fun kotlinFunction(): (String) -> String { return { it.toUpperCase() } } - + @Bean fun kotlinPojoFunction(): (Person) -> String { return { it.name.toString()} @@ -52,4 +52,5 @@ class KotlinLambdasConfiguration { fun javaFunction(): Function { return Function { x -> x } } + } diff --git a/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/KotlinSuspendFlowLambdasConfiguration.kt b/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/KotlinSuspendFlowLambdasConfiguration.kt new file mode 100644 index 000000000..074335874 --- /dev/null +++ b/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/KotlinSuspendFlowLambdasConfiguration.kt @@ -0,0 +1,64 @@ +/* + * Copyright 2012-2019 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 + * + * https://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.cloud.function.kotlin + +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.collect +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.map +import org.springframework.boot.autoconfigure.EnableAutoConfiguration +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration +import reactor.core.publisher.Flux +import java.util.function.Function + +/** + * @author Adrien Poupard + * + */ +@EnableAutoConfiguration +@Configuration +class KotlinSuspendFlowLambdasConfiguration { + + @Bean + fun kotlinFunction(): suspend (Flow) -> Flow = { flow -> + flow.map { value -> value.toUpperCase() } + } + + @Bean + fun kotlinPojoFunction(): suspend (Flow) -> Flow = { flow -> + flow.map(Person::toString) + } + + @Bean + fun kotlinConsumer(): suspend (Flow) -> Unit = { flow -> + flow.collect(::println) + } + + @Bean + fun kotlinSupplier(): suspend () -> Flow = { + flow { + emit("Hello") + } + } + + @Bean + fun javaFunction(): Function, Flux> { + return Function { x -> x } + } + +} diff --git a/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/KotlinSuspendLambdasConfiguration.kt b/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/KotlinSuspendLambdasConfiguration.kt new file mode 100644 index 000000000..106247162 --- /dev/null +++ b/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/KotlinSuspendLambdasConfiguration.kt @@ -0,0 +1,46 @@ +/* + * Copyright 2012-2019 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 + * + * https://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.cloud.function.kotlin + +import org.springframework.boot.autoconfigure.EnableAutoConfiguration +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration + +/** + * @author Adrien Poupard + * + */ +@EnableAutoConfiguration +@Configuration +class KotlinSuspendLambdasConfiguration { + + @Bean + fun kotlinFunction(): suspend (Person) -> String { + return { it.name.toString()} + } + + @Bean + fun kotlinConsumer(): suspend (String) -> Unit { + return { println(it) } + } + + @Bean + fun kotlinSupplier(): suspend () -> String { + return { "Hello" } + } + +} diff --git a/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/web/HeadersToMessageSuspendTests.kt b/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/web/HeadersToMessageSuspendTests.kt new file mode 100644 index 000000000..446f3712f --- /dev/null +++ b/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/web/HeadersToMessageSuspendTests.kt @@ -0,0 +1,108 @@ +/* + * Copyright 2012-2019 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 + * + * https://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.cloud.function.kotlin.web + +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.map +import org.assertj.core.api.Assertions +import org.junit.jupiter.api.Test +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.boot.autoconfigure.EnableAutoConfiguration +import org.springframework.boot.test.context.SpringBootTest +import org.springframework.boot.test.context.SpringBootTest.WebEnvironment +import org.springframework.boot.test.web.client.TestRestTemplate +import org.springframework.cloud.function.web.RestApplication +import org.springframework.context.annotation.Bean +import org.springframework.http.MediaType +import org.springframework.http.RequestEntity +import org.springframework.messaging.Message +import org.springframework.messaging.support.MessageBuilder +import org.springframework.test.context.ContextConfiguration +import java.net.URI + +/** + * @author Adrien Poupard + */ +@SpringBootTest( + webEnvironment = WebEnvironment.RANDOM_PORT, + properties = ["spring.cloud.function.web.path=/functions", "spring.main.web-application-type=reactive"] +) +@ContextConfiguration(classes = [RestApplication::class, HeadersToMessageSuspendTests.TestConfiguration::class]) +class HeadersToMessageSuspendTests { + @Autowired + private val rest: TestRestTemplate? = null + @Test + @Throws(Exception::class) + fun testBodyAndCustomHeaderFromMessagePropagation() { + // test POJO paylod + var postForEntity = rest!! + .exchange( + RequestEntity.post(URI("/functions/employeeSuspend")) + .contentType(MediaType.APPLICATION_JSON) + .body("[{\"name\":\"Bob\",\"age\":25}]"), String::class.java + ) + + Assertions.assertThat(postForEntity.body).isEqualTo("[{\"name\":\"Bob\",\"age\":25}]") + Assertions.assertThat(postForEntity.headers.containsKey("x-content-type")).isTrue + Assertions.assertThat(postForEntity.headers["x-content-type"]!![0]) + .isEqualTo("application/xml") + Assertions.assertThat(postForEntity.headers["foo"]!![0]).isEqualTo("bar") + + // test simple type payload + postForEntity = rest.postForEntity( + URI("/functions/stringSuspend"), + "HELLO", String::class.java + ) + Assertions.assertThat(postForEntity.body).isEqualTo("[\"HELLO\"]") + Assertions.assertThat(postForEntity.headers.containsKey("x-content-type")).isTrue + Assertions.assertThat(postForEntity.headers["x-content-type"]!![0]) + .isEqualTo("application/xml") + Assertions.assertThat(postForEntity.headers["foo"]!![0]).isEqualTo("bar") + } + + @EnableAutoConfiguration + @org.springframework.boot.test.context.TestConfiguration + class TestConfiguration { + @Bean("stringSuspend") + fun functiono():suspend (employee: Flow>) -> Flow> = { flow: Flow> -> + flow.map { request -> + val message = + MessageBuilder.withPayload(request.payload) + .setHeader("X-Content-Type", "application/xml") + .setHeader("foo", "bar").build() + message + } + } + + @Bean("employeeSuspend") + fun function1(): suspend (employee: Flow>) -> Flow> = { flow -> + flow.map { request -> + val message = + MessageBuilder + .withPayload(request.payload) + .setHeader("X-Content-Type", "application/xml") + .setHeader("foo", "bar") + .build() + message + } + } + } + + class Employee { + var name: String? = null + var age = 0 + } +} diff --git a/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/web/HeadersToMessageTests.kt b/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/web/HeadersToMessageTests.kt new file mode 100644 index 000000000..6ba600d62 --- /dev/null +++ b/spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/web/HeadersToMessageTests.kt @@ -0,0 +1,103 @@ +/* + * Copyright 2012-2019 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 + * + * https://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.cloud.function.kotlin.web + +import org.assertj.core.api.Assertions +import org.junit.jupiter.api.Test +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.boot.autoconfigure.EnableAutoConfiguration +import org.springframework.boot.test.context.SpringBootTest +import org.springframework.boot.test.context.SpringBootTest.WebEnvironment +import org.springframework.boot.test.web.client.TestRestTemplate +import org.springframework.cloud.function.web.RestApplication +import org.springframework.context.annotation.Bean +import org.springframework.http.MediaType +import org.springframework.http.RequestEntity +import org.springframework.messaging.Message +import org.springframework.messaging.support.MessageBuilder +import org.springframework.test.context.ContextConfiguration +import java.lang.Exception +import java.net.URI + +/** + * @author Dave Syer + * @author Oleg Zhurakousky + */ +@SpringBootTest( + webEnvironment = WebEnvironment.RANDOM_PORT, + properties = ["spring.cloud.function.web.path=/functions", "spring.main.web-application-type=reactive"] +) +@ContextConfiguration(classes = [RestApplication::class, HeadersToMessageTests.TestConfiguration::class]) +class HeadersToMessageTests { + @Autowired + private val rest: TestRestTemplate? = null + @Test + @Throws(Exception::class) + fun testBodyAndCustomHeaderFromMessagePropagation() { + // test POJO paylod + var postForEntity = rest!! + .exchange( + RequestEntity.post(URI("/functions/employee")) + .contentType(MediaType.APPLICATION_JSON) + .body("{\"name\":\"Bob\",\"age\":25}"), String::class.java + ) + Assertions.assertThat(postForEntity.body).isEqualTo("{\"name\":\"Bob\",\"age\":25}") + Assertions.assertThat(postForEntity.headers.containsKey("x-content-type")).isTrue + Assertions.assertThat(postForEntity.headers["x-content-type"]!![0]) + .isEqualTo("application/xml") + Assertions.assertThat(postForEntity.headers["foo"]!![0]).isEqualTo("bar") + + // test simple type payload + postForEntity = rest.postForEntity( + URI("/functions/string"), + "{\"name\":\"Bob\",\"age\":25}", String::class.java + ) + Assertions.assertThat(postForEntity.body).isEqualTo("{\"name\":\"Bob\",\"age\":25}") + Assertions.assertThat(postForEntity.headers.containsKey("x-content-type")).isTrue + Assertions.assertThat(postForEntity.headers["x-content-type"]!![0]) + .isEqualTo("application/xml") + Assertions.assertThat(postForEntity.headers["foo"]!![0]).isEqualTo("bar") + } + + @EnableAutoConfiguration + @org.springframework.boot.test.context.TestConfiguration + class TestConfiguration { + @Bean("string") + fun functiono(): (message: Message) -> Message = { request: Message -> + val message = + MessageBuilder.withPayload(request.payload) + .setHeader("X-Content-Type", "application/xml") + .setHeader("foo", "bar").build() + message + } + + @Bean("employee") + fun function1(): (employee: Message) -> Message = { request -> + val message = + MessageBuilder + .withPayload(request.payload) + .setHeader("X-Content-Type", "application/xml") + .setHeader("foo", "bar").build() + message + } + } + + // used by json converter + class Employee { + var name: String? = null + var age = 0 + } +} diff --git a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/flux/HeadersToMessageTests.java b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/flux/HeadersToMessageTests.java index 8f72efed5..45f40cb7d 100644 --- a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/flux/HeadersToMessageTests.java +++ b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/flux/HeadersToMessageTests.java @@ -25,9 +25,9 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; +import org.springframework.boot.test.context.TestConfiguration; import org.springframework.boot.test.web.client.TestRestTemplate; import org.springframework.cloud.function.web.RestApplication; -import org.springframework.cloud.function.web.flux.HeadersToMessageTests.TestConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.http.MediaType; import org.springframework.http.RequestEntity; @@ -41,12 +41,13 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Dave Syer * @author Oleg Zhurakousky + * @author Adrien Poupard * */ @SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT, properties = { "spring.cloud.function.web.path=/functions", "spring.main.web-application-type=reactive" }) -@ContextConfiguration(classes = { RestApplication.class, TestConfiguration.class }) +@ContextConfiguration(classes = { RestApplication.class, HeadersToMessageTests.TestConfiguration.class }) public class HeadersToMessageTests { @Autowired