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

GH-2027 (cont) clean up and polish

Browse Source
This commit is contained in:
Oleg Zhurakousky
2020-10-21 08:19:59 +02:00
parent ed6bf1dda4
commit ced60f548f
1 changed files with 4 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -163,6 +163,7 @@ public class FunctionConfiguration {
return new InitializingBean() {
@SuppressWarnings({ "rawtypes", "unchecked" })
@Override
public void afterPropertiesSet() throws Exception {
for (BindableFunctionProxyFactory proxyFactory : proxyFactories) {
@@ -193,10 +194,6 @@ public class FunctionConfiguration {
supplier = functionCatalog.lookup(supplierName);
function = functionCatalog.lookup(remainingFunctionDefinition, contentTypes.toArray(new String[0]));
// functionWrapper = ((FunctionInvocationWrapper) function).isInputTypePublisher()
// && ((FunctionInvocationWrapper) supplier).isOutputTypePublisher()
// ? functionCatalog.lookup(proxyFactory.getFunctionDefinition(), contentTypes.toArray(new String[0]))
// : null;
Type inputType = FunctionTypeUtils.getInputType(((FunctionInvocationWrapper) function).getFunctionType(), 0);
Type outputType = FunctionTypeUtils.getOutputType(((FunctionInvocationWrapper) supplier).getFunctionType(), 0);
if (FunctionTypeUtils.isPublisher(inputType) && !FunctionTypeUtils.isPublisher(outputType)) {
@@ -206,6 +203,9 @@ public class FunctionConfiguration {
functionWrapper = functionCatalog.lookup(proxyFactory.getFunctionDefinition(), contentTypes.toArray(new String[0]));
}
}
else {
functionWrapper = functionCatalog.lookup(proxyFactory.getFunctionDefinition(), contentTypes.toArray(new String[0]));
}
Publisher<Object> beginPublishingTrigger = setupBindingTrigger(context);
@@ -213,22 +213,6 @@ public class FunctionConfiguration {
String integrationFlowName = proxyFactory.getFunctionDefinition() + "_integrationflow";
PollableBean pollable = extractPollableAnnotation(functionProperties, context, proxyFactory);
//<<<<<<< HEAD
// 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 dynamicDestinationResolver.resolveDestination(destinationName);
// }
// 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),
@@ -262,7 +246,6 @@ public class FunctionConfiguration {
.applyBeanPostProcessorsBeforeInitialization(integrationFlow, integrationFlowName);
context.registerBean(integrationFlowName, IntegrationFlow.class, () -> postProcessedFlow);
}
//>>>>>>> a7d2e14e... GH-2027 Ensure imperative Supplier behavior during composition
}
}
}