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

GH-2303 Performance improvements

Browse Source 
Most of the work is done in s-c-function though, so here it is very light

Resolves #2303
This commit is contained in:
Oleg Zhurakousky
2022-03-22 13:56:03 +01:00
parent 95ea69f9ac
commit 1edf4db14e
2 changed files with 2 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -99,7 +99,6 @@ import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.IdGenerators.SimpleIncrementingIdGenerator;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.IntegrationReactiveUtils;
import org.springframework.lang.Nullable;
@@ -115,7 +114,6 @@ import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.IdGenerator;
import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
@@ -135,11 +133,6 @@ import org.springframework.util.StringUtils;
@ConditionalOnBean(FunctionRegistry.class)
public class FunctionConfiguration {
@Bean
public IdGenerator idGenerator() {
return new SimpleIncrementingIdGenerator();
}
@SuppressWarnings("rawtypes")
@Bean
public StreamBridge streamBridgeUtils(FunctionCatalog functionCatalog, FunctionRegistry functionRegistry,
@@ -166,9 +159,6 @@ public class FunctionConfiguration {
StreamBridge streamBridge) {
boolean shouldCreateInitializer = true;
// boolean shouldCreateInitializer = applicationContext.containsBean("output");
// || ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class));
return shouldCreateInitializer
? new FunctionToDestinationBinder(functionCatalog, functionProperties,
serviceProperties, streamBridge)
@@ -243,7 +233,6 @@ public class FunctionConfiguration {
PollableBean pollable = extractPollableAnnotation(functionProperties, context, proxyFactory);
if (functionWrapper != null) {
// Type functionType = functionWrapper.getFunctionType();
IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper(functionWrapper, context, producerProperties),
beginPublishingTrigger, pollable, context, taskScheduler, producerProperties, outputName)
.route(Message.class, message -> {

8
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -36,8 +36,6 @@ import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry
import org.springframework.cloud.function.context.message.MessageUtils;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.PartitionHandler;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.cloud.stream.binding.BindingService;
import org.springframework.cloud.stream.binding.DefaultPartitioningInterceptor;
@@ -47,10 +45,8 @@ import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.ResolvableType;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.config.GlobalChannelInterceptorProcessor;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
@@ -226,7 +222,7 @@ public final class StreamBridge implements SmartInitializingSingleton {
functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties);
}
Message<byte[]> resultMessage = (Message<byte[]>) functionToInvoke.apply(data);
return messageChannel.send(resultMessage);
}
@@ -247,7 +243,7 @@ public final class StreamBridge implements SmartInitializingSingleton {
if (this.initialized) {
return;
}
FunctionRegistration<Function<Object, Object>> fr = new FunctionRegistration<>(new PassThruFunction(), STREAM_BRIDGE_FUNC_NAME);
fr.getProperties().put("singleton", "false");
Type functionType = ResolvableType.forClassWithGenerics(Function.class, Object.class, Object.class).getType();