SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2027 Ensure imperative Supplier behavior during composition

Browse Source 
This fix ensures that whenever imperative Supplier is composed with reactive function its poling behavior and expectations are preserved

Resolves #2027
This commit is contained in:
Oleg Zhurakousky
2020-10-21 07:56:10 +02:00
parent 0c0aaec148
commit a7d2e14e80
2 changed files with 97 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -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<Object> 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<? super Flux<Message<Object>>, ? extends Publisher<Object>>) 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);
}
}
}
}

43
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
View File

@@ -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<byte[]> 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<Long> supplier() {
return () -> System.currentTimeMillis();
}
@Bean
public Function<Flux<String>, Flux<String>> functionA() {
return flux -> flux;
}
}
@EnableAutoConfiguration(exclude = ContextFunctionCatalogAutoConfiguration.class)
public static class NoFunctionEnabledConfiguration {