Fix AggregatorFunctionConfiguration for Flux race condition

Defer consumption (subscription) in the `aggregatorFunction()`
from the input channel until request is done in the output channel.
This way we ensure that output `FluxMessageChannel` got subscribers
before `subscribeTo()` starts to emit data

Author: artembilan

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

@@ -79,9 +79,7 @@ public Function<Flux<Message<?>>, Flux<Message<?>>> aggregatorFunction(
 				.map((inputMessage) -> MessageBuilder.fromMessage(inputMessage).removeHeader("kafka_consumer").build())
 				.delaySubscription(Duration.ZERO);
 
-			aggregatorInputChannel.subscribeTo(messageFlux);
-
-			return Flux.from(this.outputChannel);
+			return Flux.from(this.outputChannel).doOnRequest((__) -> aggregatorInputChannel.subscribeTo(messageFlux));
 		};
 	}