SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-112: Fix JdbcConsumerConfiguration for auto-wire ambiguity

Browse Source 
Fixes: https://github.com/spring-cloud/spring-functions-catalog/issues/112
This commit is contained in:
Artem Bilan
2024-12-19 11:55:55 -05:00
parent a0d01bf462
commit d5a77d1399
1 changed files with 7 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
consumer/spring-jdbc-consumer/src/main/java/org/springframework/cloud/fn/consumer/jdbc/JdbcConsumerConfiguration.java
View File

@@ -51,7 +51,6 @@ import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.gateway.AnnotationGatewayProxyFactoryBean;
import org.springframework.integration.jdbc.JdbcMessageHandler;
import org.springframework.integration.jdbc.SqlParameterSourceFactory;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.MutableMessage;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
@@ -120,8 +119,8 @@ public class JdbcConsumerConfiguration {
}
@Bean
IntegrationFlow jdbcConsumerFlow(@Qualifier("aggregator") MessageHandler aggregator,
JdbcMessageHandler jdbcMessageHandler) {
IntegrationFlow jdbcConsumerFlow(@Qualifier("jdbcConsumerAggregator") MessageHandler aggregator,
@Qualifier("jdbcConsumerMessageHandler") JdbcMessageHandler jdbcMessageHandler) {
return (flow) -> {
if (this.properties.getBatchSize() > 1 || this.properties.getIdleTimeout() > 0) {
@@ -140,13 +139,16 @@ public class JdbcConsumerConfiguration {
}
@Bean
FactoryBean<MessageHandler> aggregator(MessageGroupStore messageGroupStore) {
FactoryBean<MessageHandler> jdbcConsumerAggregator() {
AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
aggregatorFactoryBean.setCorrelationStrategy((message) -> message.getPayload().getClass().getName());
aggregatorFactoryBean.setReleaseStrategy(new MessageCountReleaseStrategy(this.properties.getBatchSize()));
if (this.properties.getIdleTimeout() >= 0) {
aggregatorFactoryBean.setGroupTimeoutExpression(new ValueExpression<>(this.properties.getIdleTimeout()));
}
SimpleMessageStore messageGroupStore = new SimpleMessageStore();
messageGroupStore.setTimeoutOnIdle(true);
messageGroupStore.setCopyOnGet(false);
aggregatorFactoryBean.setMessageStore(messageGroupStore);
aggregatorFactoryBean.setProcessorBean(new DefaultAggregatingMessageGroupProcessor());
aggregatorFactoryBean.setExpireGroupsUponCompletion(true);
@@ -155,15 +157,7 @@ public class JdbcConsumerConfiguration {
}
@Bean
MessageGroupStore messageGroupStore() {
SimpleMessageStore messageGroupStore = new SimpleMessageStore();
messageGroupStore.setTimeoutOnIdle(true);
messageGroupStore.setCopyOnGet(false);
return messageGroupStore;
}
@Bean
public JdbcMessageHandler jdbcMessageHandler(DataSource dataSource,
public JdbcMessageHandler jdbcConsumerMessageHandler(DataSource dataSource,
@Qualifier(IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME) EvaluationContext evaluationContext) {
final MultiValueMap<String, Expression> columnExpressionVariations = new LinkedMultiValueMap<>();