From 5be4226fae35f1d1ba2a892d7d5156de4272c75c Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 19 Jan 2022 11:09:53 +0100 Subject: [PATCH] Upgrade function version and address removal of certain classes in function --- pom.xml | 3 +-- .../cloud/stream/function/FunctionConfiguration.java | 6 ++++-- .../springframework/cloud/stream/function/StreamBridge.java | 6 ++++-- .../cloud/stream/function/ImplicitFunctionBindingTests.java | 4 ++-- 4 files changed, 11 insertions(+), 8 deletions(-) diff --git a/pom.xml b/pom.xml index 7aeabee1f..494cffc2b 100644 --- a/pom.xml +++ b/pom.xml @@ -4,7 +4,6 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0 spring-cloud-stream-parent - 4.0.0-SNAPSHOT pom org.springframework.cloud @@ -25,7 +24,7 @@ 17 2020.0.7 2.1 - 3.2.2-SNAPSHOT + 4.0.0-SNAPSHOT true true true diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index e5205d9a3..ae87ef2f1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -58,7 +58,6 @@ import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.FunctionProperties; import org.springframework.cloud.function.context.FunctionRegistry; import org.springframework.cloud.function.context.PollableBean; -import org.springframework.cloud.function.context.catalog.FunctionInspector; import org.springframework.cloud.function.context.catalog.FunctionTypeUtils; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration; @@ -939,7 +938,10 @@ public class FunctionConfiguration { streamFunctionProperties.setDefinition(RoutingFunction.FUNCTION_NAME); } else if (autodetect) { - streamFunctionProperties.setDefinition(((FunctionInspector) functionCatalog).getName(functionCatalog.lookup(""))); + FunctionInvocationWrapper function = functionCatalog.lookup(""); + if (function != null) { + streamFunctionProperties.setDefinition(function.getFunctionDefinition()); + } } return StringUtils.hasText(streamFunctionProperties.getDefinition()); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 2d259e07b..2c2dbeecd 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.function; +import java.lang.reflect.Type; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; @@ -29,7 +30,6 @@ import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.FunctionRegistration; import org.springframework.cloud.function.context.FunctionRegistry; -import org.springframework.cloud.function.context.FunctionType; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.function.context.message.MessageUtils; import org.springframework.cloud.stream.binder.Binder; @@ -42,6 +42,7 @@ import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel; import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.core.ResolvableType; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.config.GlobalChannelInterceptorProcessor; import org.springframework.integration.support.MessageBuilder; @@ -241,7 +242,8 @@ public final class StreamBridge implements SmartInitializingSingleton { } FunctionRegistration> fr = new FunctionRegistration<>(v -> v, STREAM_BRIDGE_FUNC_NAME); fr.getProperties().put("singleton", "false"); - this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message())); + Type functionType = ResolvableType.forClassWithGenerics(Function.class, Object.class, Object.class).getType(); + this.functionRegistry.register(fr.type(functionType)); Map channels = applicationContext.getBeansOfType(DirectWithAttributesChannel.class); for (Entry channelEntry : channels.entrySet()) { if (channelEntry.getValue().getAttribute("type").equals("output")) { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java index 7f225cc57..4e1e5f015 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java @@ -41,7 +41,6 @@ import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.function.context.FunctionRegistration; -import org.springframework.cloud.function.context.FunctionType; import org.springframework.cloud.function.context.catalog.FunctionAroundWrapper; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration; @@ -57,6 +56,7 @@ import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel; import org.springframework.context.ApplicationListener; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.core.ResolvableType; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; @@ -157,7 +157,7 @@ public class ImplicitFunctionBindingTests { Function function = v -> v; FunctionRegistration functionRegistration = new FunctionRegistration(function, "function"); - functionRegistration = functionRegistration.type(FunctionType.from(byte[].class).to(byte[].class)); + functionRegistration = functionRegistration.type(ResolvableType.forClassWithGenerics(Function.class, Object.class, Object.class).getType()); FunctionBindingTestUtils.bind(context, functionRegistration); input.send(new GenericMessage("hello".getBytes()), "input");