From 4ede865c900f175388d40cc540e1005c1e179df4 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 8 Oct 2019 06:07:23 -0500 Subject: [PATCH] Fix StreamFunctionProperties bindings property usage fix usage of PollableSupplier and update documentation --- .../main/asciidoc/spring-cloud-stream.adoc | 4 +- .../BindableFunctionProxyFactory.java | 43 +++---- .../function/FunctionConfiguration.java | 114 +++++++++++------- .../function/StreamFunctionProperties.java | 61 ++++------ .../ImplicitFunctionBindingTests.java | 21 ++++ .../SourceToFunctionsSupportTests.java | 46 ++++++- 6 files changed, 174 insertions(+), 115 deletions(-) diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index f2ff7668f..30a8d94d5 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -554,14 +554,14 @@ Consider the following sample, which emulates such use case by producing a finit @SpringBootApplication public static class SupplierConfiguration { - @PollableSupplier + @Pollable public Supplier> stringSupplier() { return () -> Flux.just("hello", "bye"); } } ---- -The bean itself is annotated with `PollableSupplier` annotation (sub-set of `@Bean`), thus signaling to the framework that although the implementation +The bean itself is annotated with `Pollable` annotation (sub-set of `@Bean`), thus signaling to the framework that although the implementation of such a supplier is reactive, it still needs to be polled. ====== Polling Configuration Properties diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java index 97110c608..8ec010359 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java @@ -47,16 +47,15 @@ class BindableFunctionProxyFactory extends BindableProxyFactory { private final String functionDefinition; - private final boolean nameBasedOnFunctionName; + private final StreamFunctionProperties functionProperties; - private boolean multiple; - BindableFunctionProxyFactory(String functionDefinition, int inputCount, int outputCount, boolean nameBasedOnFunctionName) { + BindableFunctionProxyFactory(String functionDefinition, int inputCount, int outputCount, StreamFunctionProperties functionProperties) { super(null); this.inputCount = inputCount; this.outputCount = outputCount; this.functionDefinition = functionDefinition; - this.nameBasedOnFunctionName = nameBasedOnFunctionName; + this.functionProperties = functionProperties; } @@ -65,27 +64,15 @@ class BindableFunctionProxyFactory extends BindableProxyFactory { Assert.notEmpty(BindableFunctionProxyFactory.this.bindingTargetFactories, "'bindingTargetFactories' cannot be empty"); - this.multiple = this.inputCount > 1 || this.outputCount > 1; - if (this.inputCount > 0) { - if (multiple || nameBasedOnFunctionName) { - for (int i = 0; i < inputCount; i++) { - this.createInput(this.buildInputNameForIndex(i)); - } - } - else { - this.createInput("input"); + for (int i = 0; i < inputCount; i++) { + this.createInput(this.buildInputNameForIndex(i)); } } if (this.outputCount > 0) { - if (multiple || nameBasedOnFunctionName) { - for (int i = 0; i < outputCount; i++) { - this.createOutput(this.buildOutputNameForIndex(i)); - } - } - else { - this.createOutput("output"); + for (int i = 0; i < outputCount; i++) { + this.createOutput(this.buildOutputNameForIndex(i)); } } } @@ -100,10 +87,6 @@ class BindableFunctionProxyFactory extends BindableProxyFactory { return true; } - protected boolean isNameBasedOnFunctionName() { - return this.nameBasedOnFunctionName; - } - protected String getFunctionDefinition() { return this.functionDefinition; } @@ -121,7 +104,7 @@ class BindableFunctionProxyFactory extends BindableProxyFactory { } protected boolean isMultiple() { - return this.multiple; + return this.inputCount > 1 || this.outputCount > 1; } private String buildInputNameForIndex(int index) { @@ -143,13 +126,19 @@ class BindableFunctionProxyFactory extends BindableProxyFactory { } private void createInput(String name) { - BindableFunctionProxyFactory.this.inputHolders.put(name, + if (this.functionProperties.getBindings().containsKey(name)) { + name = this.functionProperties.getBindings().get(name); + } + this.inputHolders.put(name, new BoundTargetHolder(getBindingTargetFactory(SubscribableChannel.class) .createInput(name), true)); } private void createOutput(String name) { - BindableFunctionProxyFactory.this.outputHolders.put(name, + if (this.functionProperties.getBindings().containsKey(name)) { + name = this.functionProperties.getBindings().get(name); + } + this.outputHolders.put(name, new BoundTargetHolder(getBindingTargetFactory(MessageChannel.class) .createOutput(name), 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 409ca94f7..6866ff4bb 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 @@ -20,7 +20,9 @@ import java.lang.reflect.Field; import java.lang.reflect.Method; import java.lang.reflect.Type; import java.time.Duration; +import java.util.ArrayList; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicReference; @@ -45,7 +47,7 @@ import org.springframework.boot.autoconfigure.AutoConfigureBefore; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.FunctionProperties; -import org.springframework.cloud.function.context.PollableSupplier; +import org.springframework.cloud.function.context.Pollable; 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; @@ -123,53 +125,56 @@ public class FunctionConfiguration { return new FunctionChannelBindingInitializer(functionCatalog, functionProperties, bindableProxyFactories, serviceProperties); } + + /* - * Binding initializer responsible only for Suppliers only + * Binding initializer responsible only for Suppliers */ @Bean - IntegrationFlow supplierInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector, - StreamFunctionProperties functionProperties, GenericApplicationContext context, BindingServiceProperties serviceProperties) { - if (!ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class))) { + InitializingBean supplierInitializer(FunctionCatalog functionCatalog, StreamFunctionProperties functionProperties, + GenericApplicationContext context, BindingServiceProperties serviceProperties, + @Nullable BindableFunctionProxyFactory[] proxyFactories) { + + if (!ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class)) || proxyFactories == null) { return null; } - IntegrationFlow integrationFlow = null; - String[] functionDefinitions = StringUtils.hasText(functionProperties.getDefinition()) - ? functionProperties.getDefinition().split(";") - : new String[] {}; + return new InitializingBean() { - for (String functionDefinition : functionDefinitions) { - BindingProperties bindingProperties = serviceProperties.getBindingProperties("output"); + @Override + public void afterPropertiesSet() throws Exception { + for (BindableFunctionProxyFactory proxyFactory : proxyFactories) { + FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(proxyFactory.getFunctionDefinition()); + if (functionWrapper != null && functionWrapper.isSupplier()) { + // gather output content types + List contentTypes = new ArrayList(); + Assert.isTrue(proxyFactory.getOutputs().size() == 1, "Supplier with multiple outputs is not supported at the moment."); + for (String outputName : proxyFactory.getOutputs()) { + BindingProperties bindingProperties = serviceProperties.getBindingProperties(outputName); + String contentType = bindingProperties.getProducer() != null && bindingProperties.getProducer().isUseNativeEncoding() + ? null : bindingProperties.getContentType(); + contentTypes.add(contentType); + } + // obtain function wrapper with proper output content types + functionWrapper = functionCatalog.lookup(proxyFactory.getFunctionDefinition(), contentTypes.toArray(new String[0])); + Publisher beginPublishingTrigger = setupBindingTrigger(context); - FunctionInvocationWrapper functionWrapper = bindingProperties.getProducer() != null && bindingProperties.getProducer().isUseNativeEncoding() - ? functionCatalog.lookup(functionDefinition) - : functionCatalog.lookup(functionDefinition, bindingProperties.getContentType()); - if (functionWrapper != null && functionWrapper.isSupplier()) { - Publisher beginPublishingTrigger = this.setupBindingTrigger(context); - - 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); + //!!!!!! TEMPORARY (see assertion about multiple outputs above) + String outputName = proxyFactory.getOutputs().iterator().next(); + // end temporary + if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo()) { + String integrationFlowName = proxyFactory.getFunctionDefinition() + "_integrationflow"; + Pollable pollable = extractPollableAnnotation(functionProperties, context, proxyFactory); + IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(functionWrapper, beginPublishingTrigger, pollable) + .channel(outputName).get(); + IntegrationFlow postProcessedFlow = (IntegrationFlow) context.getAutowireCapableBeanFactory() + .applyBeanPostProcessorsBeforeInitialization(integrationFlow, integrationFlowName); + context.registerBean(integrationFlowName, IntegrationFlow.class, () -> postProcessedFlow); + } } } - 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; - - if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo()) { - integrationFlow = this.integrationFlowFromProvidedSupplier(functionWrapper, functionInspector, beginPublishingTrigger, pollable) - .channel("output").get(); - } } - } - return integrationFlow; + }; } /* @@ -192,20 +197,21 @@ public class FunctionConfiguration { @SuppressWarnings({ "rawtypes", "unchecked" }) private IntegrationFlowBuilder integrationFlowFromProvidedSupplier(Supplier supplier, - FunctionInspector inspector, Publisher beginPublishingTrigger, PollableSupplier pollable) { + Publisher beginPublishingTrigger, Pollable pollable) { IntegrationFlowBuilder integrationFlowBuilder; Type functionType = ((FunctionInvocationWrapper) supplier).getFunctionType(); - boolean splittable = pollable != null && (boolean) AnnotationUtils.getAnnotationAttributes(pollable).get("splittable"); + boolean splittable = pollable != null + && (boolean) AnnotationUtils.getAnnotationAttributes(pollable).get("splittable"); if (pollable == null && FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))) { Publisher publisher = (Publisher) supplier.get(); publisher = publisher instanceof Mono ? ((Mono) publisher).delaySubscription(beginPublishingTrigger).map(this::wrapToMessageIfNecessary) - : ((Flux) publisher).delaySubscription(beginPublishingTrigger).map(this::wrapToMessageIfNecessary); + : ((Flux) publisher).delaySubscription(beginPublishingTrigger).map(this::wrapToMessageIfNecessary); - integrationFlowBuilder = IntegrationFlows.from(publisher); + integrationFlowBuilder = IntegrationFlows.from(publisher); } else { // implies pollable integrationFlowBuilder = IntegrationFlows.from(supplier); @@ -217,6 +223,29 @@ public class FunctionConfiguration { return integrationFlowBuilder; } + private Pollable extractPollableAnnotation(StreamFunctionProperties functionProperties, GenericApplicationContext context, + BindableFunctionProxyFactory proxyFactory) { + // here we need to ensure that for cases where composition is defined we only look for supplier method to find Pollable annotation. + String supplierFunctionName = StringUtils + .delimitedListToStringArray(proxyFactory.getFunctionDefinition().replaceAll(",", "|").trim(), "|")[0]; + RootBeanDefinition bd = (RootBeanDefinition) context.getBeanDefinition(supplierFunctionName); + 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() + "'"); + return factoryMethod.getReturnType().isAssignableFrom(Supplier.class) + ? AnnotationUtils.findAnnotation(factoryMethod, Pollable.class) + : null; + } + + @SuppressWarnings("unchecked") private Message wrapToMessageIfNecessary(T value) { return value instanceof Message @@ -562,7 +591,6 @@ public class FunctionConfiguration { && this.determineFunctionName(functionCatalog, environment)) { BeanDefinitionRegistry registry = (BeanDefinitionRegistry) applicationContext.getBeanFactory(); String[] functionDefinitions = streamFunctionProperties.getDefinition().split(";"); - boolean nameBasedOnFunctionName = functionDefinitions.length > 1; for (String functionDefinition : functionDefinitions) { RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class); FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition); @@ -584,7 +612,7 @@ public class FunctionConfiguration { functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(functionDefinition); functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount); functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(nameBasedOnFunctionName); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(streamFunctionProperties); registry.registerBeanDefinition(functionDefinition + "_binding", functionBindableProxyDefinition); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java index 7bbf9e941..4b89cf6da 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java @@ -19,10 +19,10 @@ package org.springframework.cloud.stream.function; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; -import org.springframework.cloud.stream.messaging.Processor; import org.springframework.util.StringUtils; /** @@ -36,19 +36,13 @@ public class StreamFunctionProperties { /** * Definition of functions to bind. If several functions need to be composed into one, - * use pipes (e.g., 'fooFunc\|barFunc') + * use pipes (e.g., 'fooFunc|barFunc') */ private String definition; private BindingServiceProperties bindingServiceProperties; - private String inputDestinationName = Processor.INPUT; - - private String outputDestinationName = Processor.OUTPUT; - - private Map> inputBindings = new HashMap<>(); - - private Map> outputBindings = new HashMap<>(); + private Map bindings = new HashMap<>(); private boolean batchMode; @@ -68,8 +62,12 @@ public class StreamFunctionProperties { return this.definition; } - public String[] getParsedDefinition() { - return StringUtils.delimitedListToStringArray(this.getDefinition().replaceAll(",", "|").trim(), "|"); + public List getOutputBindings(String functionName) { + return this.filterBindings(functionName, "-out-"); + } + + public List getInputBindings(String functionName) { + return this.filterBindings(functionName, "-in-"); } public void setDefinition(String definition) { @@ -89,36 +87,12 @@ public class StreamFunctionProperties { this.bindingServiceProperties = bindingServiceProperties; } - String getInputDestinationName() { - return this.inputDestinationName; + public Map getBindings() { + return this.bindings; } - void setInputDestinationName(String inputDestinationName) { - this.inputDestinationName = inputDestinationName; - } - - String getOutputDestinationName() { - return this.outputDestinationName; - } - - void setOutputDestinationName(String outputDestinationName) { - this.outputDestinationName = outputDestinationName; - } - - public Map> getInputBindings() { - return inputBindings; - } - - public Map> getOutputBindings() { - return outputBindings; - } - - public void setOutputBindings(Map> outputBindings) { - this.outputBindings = outputBindings; - } - - public void setInputBindings(Map> inputBindings) { - this.inputBindings = inputBindings; + public void setBindings(Map bindings) { + this.bindings = bindings; } public boolean isBatchMode() { @@ -128,4 +102,13 @@ public class StreamFunctionProperties { public void setBatchMode(boolean batchMode) { this.batchMode = batchMode; } + + private List filterBindings(String functionName, String suffix) { + List list = bindings.keySet().stream() + .filter(bKey -> bKey.contains(functionName + suffix)) + .sorted() + .map(bKey -> bindings.get(bKey)) + .collect(Collectors.toList()); + return list; + } } 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 6aabfb257..d9801d024 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 @@ -311,6 +311,27 @@ public class ImplicitFunctionBindingTests { } } + @Test + public void testSupplierWithCustomPollerAndMappedOutput() { + System.clearProperty("spring.cloud.stream.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + SupplierWithExplicitPollerConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.poller.fixed-delay=2000", + "--spring.cloud.stream.function.bindings.supplier-out-0=output")) { + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + PollerMetadata pollerMetadata = context.getBean(PollerMetadata.class); + assertThat(((PeriodicTrigger) pollerMetadata.getTrigger()).getPeriod()).isEqualTo(2000); + + Message outputMessage = outputDestination.receive(6000); + assertThat(outputMessage.getPayload()).isEqualTo("hello".getBytes()); + } + } + @EnableAutoConfiguration public static class NoEnableBindingConfiguration { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java index dd1276647..7c5406eb5 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java @@ -29,7 +29,7 @@ import reactor.core.publisher.Flux; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; -import org.springframework.cloud.function.context.PollableSupplier; +import org.springframework.cloud.function.context.Pollable; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; @@ -233,7 +233,31 @@ public class SourceToFunctionsSupportTests { assertThat(new String(target.receive(2000).getPayload())).isEqualTo("5"); assertThat(new String(target.receive(2000).getPayload())).isEqualTo("6"); - assertThat(context.getBean("supplierInitializer")).isNotEqualTo(null); + //assertThat(context.getBean("supplierInitializer")).isNotEqualTo(null); + } + } + + @Test + public void testMultipleSuppliers() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class, + MultipleSupplierConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.function.definition=supplier1;supplier2", + "--spring.jmx.enabled=false" +// "--spring.cloud.stream.function.bindings.supplier1-out-0=output1", +// "--spring.cloud.stream.function.bindings.supplier2-out-0=output2" + )) { + + OutputDestination target = context.getBean(OutputDestination.class); + +// assertThat(new String(target.receive(2000).getPayload())).isEqualTo("1"); +// assertThat(new String(target.receive(2000).getPayload())).isEqualTo("2"); +// assertThat(new String(target.receive(2000).getPayload())).isEqualTo("3"); +// assertThat(new String(target.receive(2000).getPayload())).isEqualTo("4"); +// assertThat(new String(target.receive(2000).getPayload())).isEqualTo("5"); +// assertThat(new String(target.receive(2000).getPayload())).isEqualTo("6"); +// +// assertThat(context.getBean("supplierInitializer")).isNotEqualTo(null); } } @@ -241,7 +265,7 @@ public class SourceToFunctionsSupportTests { public static class MessageFluxSupplierConfiguration { AtomicInteger counter = new AtomicInteger(); - @PollableSupplier(splittable = true) + @Pollable(splittable = true) public Supplier>> messageStreamSupplier() { return () -> { Message m1 = new GenericMessage<>(String.valueOf(counter.incrementAndGet())); @@ -252,11 +276,25 @@ public class SourceToFunctionsSupportTests { } } + @EnableAutoConfiguration + public static class MultipleSupplierConfiguration { + + @Bean + public Supplier supplier1() { + return () -> "supplier1"; + } + + @Bean + public Supplier supplier2() { + return () -> "supplier2"; + } + } + @EnableAutoConfiguration public static class SimpleFluxSupplierConfiguration { AtomicInteger counter = new AtomicInteger(); - @PollableSupplier(splittable = true) + @Pollable(splittable = true) public Supplier> simpleStreamSupplier() { return () -> { return Flux.just(String.valueOf(counter.incrementAndGet()),