From a7d2e14e80b586119cf661650b273e29e1263142 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 21 Oct 2020 07:56:10 +0200 Subject: [PATCH] GH-2027 Ensure imperative Supplier behavior during composition This fix ensures that whenever imperative Supplier is composed with reactive function its poling behavior and expectations are preserved Resolves #2027 --- .../function/FunctionConfiguration.java | 69 +++++++++++++++---- .../ImplicitFunctionBindingTests.java | 43 ++++++++++++ 2 files changed, 97 insertions(+), 15 deletions(-) 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 45f1b844d..7f7938fe0 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 @@ -23,6 +23,7 @@ import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.time.Instant; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.Iterator; import java.util.List; @@ -186,27 +187,65 @@ public class FunctionConfiguration { contentTypes.add(bindingProperties.getContentType()); } - // obtain function wrapper with proper output content types - functionWrapper = functionCatalog.lookup(proxyFactory.getFunctionDefinition(), contentTypes.toArray(new String[0])); + // see https://github.com/spring-cloud/spring-cloud-stream/issues/2027 + String functionDefinition = proxyFactory.getFunctionDefinition(); + String[] functionNames = StringUtils.delimitedListToStringArray(functionDefinition.replaceAll(",", "|").trim(), "|"); + + Function supplier = null; + Function function = null; + if (!ObjectUtils.isEmpty(functionNames) && functionNames.length > 1) { + String supplierName = functionNames[0]; + String remainingFunctionDefinition = StringUtils + .arrayToCommaDelimitedString(Arrays.copyOfRange(functionNames, 1, functionNames.length)); + + supplier = functionCatalog.lookup(supplierName); + function = functionCatalog.lookup(remainingFunctionDefinition); + + functionWrapper = ((FunctionInvocationWrapper) function).isInputTypePublisher() + && ((FunctionInvocationWrapper) supplier).isOutputTypePublisher() + ? functionCatalog.lookup(proxyFactory.getFunctionDefinition(), contentTypes.toArray(new String[0])) + : null; + } + Publisher beginPublishingTrigger = setupBindingTrigger(context); if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo()) { String integrationFlowName = proxyFactory.getFunctionDefinition() + "_integrationflow"; PollableBean pollable = extractPollableAnnotation(functionProperties, context, proxyFactory); - Type functionType = functionWrapper.getFunctionType(); - IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper(functionWrapper, context, producerProperties), - beginPublishingTrigger, pollable, context, taskScheduler, functionType) - .route(Message.class, message -> { - if (message.getHeaders().get("spring.cloud.stream.sendto.destination") != null) { - String destinationName = (String) message.getHeaders().get("spring.cloud.stream.sendto.destination"); - return streamBridge.resolveDestination(destinationName, producerProperties); - } - return outputName; - }).get(); - IntegrationFlow postProcessedFlow = (IntegrationFlow) context.getAutowireCapableBeanFactory() - .applyBeanPostProcessorsBeforeInitialization(integrationFlow, integrationFlowName); - context.registerBean(integrationFlowName, IntegrationFlow.class, () -> postProcessedFlow); + if (functionWrapper != null) { + Type functionType = functionWrapper.getFunctionType(); + IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper(functionWrapper, context, producerProperties), + beginPublishingTrigger, pollable, context, taskScheduler, functionType) + .route(Message.class, message -> { + if (message.getHeaders().get("spring.cloud.stream.sendto.destination") != null) { + String destinationName = (String) message.getHeaders().get("spring.cloud.stream.sendto.destination"); + return streamBridge.resolveDestination(destinationName, producerProperties); + } + return outputName; + }).get(); + IntegrationFlow postProcessedFlow = (IntegrationFlow) context.getAutowireCapableBeanFactory() + .applyBeanPostProcessorsBeforeInitialization(integrationFlow, integrationFlowName); + context.registerBean(integrationFlowName, IntegrationFlow.class, () -> postProcessedFlow); + } + else { + Type functionType = ((FunctionInvocationWrapper) supplier).getFunctionType(); + IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper((FunctionInvocationWrapper) supplier, context, producerProperties), + beginPublishingTrigger, pollable, context, taskScheduler, functionType) + .channel(c -> c.direct()) + .fluxTransform((Function>, ? extends Publisher>) function) + .route(Message.class, message -> { + if (message.getHeaders().get("spring.cloud.stream.sendto.destination") != null) { + String destinationName = (String) message.getHeaders().get("spring.cloud.stream.sendto.destination"); + return streamBridge.resolveDestination(destinationName, producerProperties); + } + return outputName; + }) + .get(); + IntegrationFlow postProcessedFlow = (IntegrationFlow) context.getAutowireCapableBeanFactory() + .applyBeanPostProcessorsBeforeInitialization(integrationFlow, integrationFlowName); + context.registerBean(integrationFlowName, IntegrationFlow.class, () -> postProcessedFlow); + } } } } 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 4ca3fc0af..072318fe5 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 @@ -489,6 +489,34 @@ public class ImplicitFunctionBindingTests { } } + @Test + public void testImperativeSupplierReactiveFunctionComposition() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(ImperativeSupplierComposedWithReactiveFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.poller.fixed-delay=1000", + "--spring.cloud.function.definition=supplier|functionA")) { + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message outputMessage = outputDestination.receive(2000); + Long value = Long.parseLong(new String(outputMessage.getPayload())); + + outputMessage = outputDestination.receive(5000); + assertThat(Long.parseLong(new String(outputMessage.getPayload())) - value).isGreaterThanOrEqualTo(1000); + + outputMessage = outputDestination.receive(5000); + assertThat(Long.parseLong(new String(outputMessage.getPayload())) - value).isGreaterThanOrEqualTo(1000); + + outputMessage = outputDestination.receive(5000); + assertThat(Long.parseLong(new String(outputMessage.getPayload())) - value).isGreaterThanOrEqualTo(1000); + } + } + + + @Test public void testSupplierWithCustomPollerAndMappedOutput() { System.clearProperty("spring.cloud.function.definition"); @@ -1089,6 +1117,21 @@ public class ImplicitFunctionBindingTests { } } + @EnableAutoConfiguration + public static class ImperativeSupplierComposedWithReactiveFunctionConfiguration { + + @Bean + public Supplier supplier() { + return () -> System.currentTimeMillis(); + } + + @Bean + public Function, Flux> functionA() { + return flux -> flux; + } + + } + @EnableAutoConfiguration(exclude = ContextFunctionCatalogAutoConfiguration.class) public static class NoFunctionEnabledConfiguration {