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

Upgrade to Spring Boot 3.2.7

Browse Source 
Fix `AggregatorFunctionConfiguration` according to a new default logic in the `FluxMessageChannel`
This commit is contained in:
Artem Bilan
2024-06-24 13:12:30 -04:00
parent b30e3ced21
commit bd2e7fc741
2 changed files with 5 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
function/spring-aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java
View File

@@ -19,6 +19,7 @@ package org.springframework.cloud.fn.aggregator;
import java.util.function.Function;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
@@ -67,8 +68,9 @@ public class AggregatorFunctionConfiguration {
@Bean
public Function<Flux<Message<?>>, Flux<Message<?>>> aggregatorFunction(FluxMessageChannel aggregatorInputChannel) {
return (input) -> Flux.from(this.outputChannel)
.doOnRequest((request) -> aggregatorInputChannel.subscribeTo(input.map((
inputMessage) -> MessageBuilder.fromMessage(inputMessage).removeHeader("kafka_consumer").build())));
.doOnRequest((request) -> aggregatorInputChannel.subscribeTo(input
.map((inputMessage) -> MessageBuilder.fromMessage(inputMessage).removeHeader("kafka_consumer").build())
.publishOn(Schedulers.boundedElastic())));
}
@Bean