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 5a8b08b81..a7ed72d94 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 @@ -41,6 +41,7 @@ import org.springframework.cloud.function.context.PollableSupplier; import org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.function.context.catalog.FunctionInspector; import org.springframework.cloud.function.context.catalog.FunctionTypeUtils; +import org.springframework.cloud.function.context.config.FunctionContextUtils; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.BindingCreatedEvent; import org.springframework.cloud.stream.binding.BindableProxyFactory; @@ -54,6 +55,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.context.support.GenericApplicationContext; import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.core.type.MethodMetadata; import org.springframework.integration.channel.MessageChannelReactiveUtils; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlowBuilder; @@ -66,8 +68,10 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.SubscribableChannel; import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; import org.springframework.util.MimeTypeUtils; import org.springframework.util.ObjectUtils; +import org.springframework.util.ReflectionUtils; /** * @author Oleg Zhurakousky @@ -95,7 +99,7 @@ public class FunctionConfiguration { IntegrationFlow integrationFlow = null; if (functionCatalog != null && ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class))) { FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(functionProperties.getDefinition()); - if (functionWrapper != null) { + if (functionWrapper != null /*&& functionWrapper.getTarget() instanceof Supplier*/) { AtomicReference> triggerRef = new AtomicReference<>(); Publisher beginPublishingTrigger = Mono.create(emmiter -> { triggerRef.set(emmiter); @@ -108,9 +112,18 @@ public class FunctionConfiguration { } }); - RootBeanDefinition bd = (RootBeanDefinition) context.getBeanDefinition(functionProperties.getParsedDefinition()[0]); Method factoryMethod = bd.getResolvedFactoryMethod(); + if (factoryMethod == null) { + Object source = bd.getSource(); + if (source instanceof MethodMetadata) { + Class factory = ClassUtils.resolveClassName(((MethodMetadata) source).getDeclaringClassName(), null); + Class[] params = FunctionContextUtils.getParamTypesFromBeanDefinitionFactory(factory, bd); + factoryMethod = ReflectionUtils.findMethod(factory, ((MethodMetadata) source).getMethodName(), params); + } + } + Assert.notNull(factoryMethod, "Failed to introspect factory method since it was not discovered for function '" + + functionProperties.getDefinition() + "'"); PollableSupplier pollable = factoryMethod.getReturnType().isAssignableFrom(Supplier.class) ? AnnotationUtils.findAnnotation(factoryMethod, PollableSupplier.class) : null; @@ -158,7 +171,6 @@ public class FunctionConfiguration { return value instanceof Message ? (Message) value : MessageBuilder.withPayload(value).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build(); } - /** * * @author Oleg Zhurakousky diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/edgecases/GH1801Test.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/edgecases/GH1801Test.java new file mode 100644 index 000000000..4ec03c009 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/edgecases/GH1801Test.java @@ -0,0 +1,59 @@ +/* + * Copyright 2019-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.stream.function.edgecases; + +import java.util.function.Consumer; + +import org.junit.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.Message; + +/** + * This test validates that the issue https://github.com/spring-cloud/spring-cloud-stream/issues/1801 + * is addressed. + * + * @author Oleg Zhurakousky + * + */ +public class GH1801Test { + + @Test + public void test() { + SampleBootApplication.main("--spring.cloud.stream.defaultBinder=integration"); + } + + @SpringBootApplication + public static class SampleBootApplication { + public static void main(String... args) { + new SpringApplicationBuilder(SampleBootApplication.class).web(WebApplicationType.NONE).run(args); + } + } +} + +@Configuration +class StreamConfiguration { + + @Bean + public Consumer> consumer() { + return System.out::println; + } +}