Prefix function names with spring-

* Add `spring-` prefix to function names

Fixes: #9

This commit renames each sub-module in the common, consumer, function
and supplier groups with a prefix of `spring-`.

* Update README.adoc links to new prefixed names

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

@@ -0,0 +1,164 @@
+/*
+ * Copyright 2020-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.aggregator;
+
+import java.util.function.Function;
+
+import reactor.core.publisher.Flux;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.fn.common.config.ComponentCustomizer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.aggregator.CorrelationStrategy;
+import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor;
+import org.springframework.integration.aggregator.ExpressionEvaluatingCorrelationStrategy;
+import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor;
+import org.springframework.integration.aggregator.ExpressionEvaluatingReleaseStrategy;
+import org.springframework.integration.aggregator.MessageGroupProcessor;
+import org.springframework.integration.aggregator.ReleaseStrategy;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.integration.config.AggregatorFactoryBean;
+import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Artem Bilan
+ * @author Corneil du Plessis
+ */
+@AutoConfiguration
+@EnableConfigurationProperties(AggregatorFunctionProperties.class)
+public class AggregatorFunctionConfiguration {
+
+	@Autowired
+	private AggregatorFunctionProperties properties;
+
+	@Autowired
+	private BeanFactory beanFactory;
+
+	@Bean
+	public Function<Flux<Message<?>>, Flux<Message<?>>> aggregatorFunction(
+		FluxMessageChannel inputChannel,
+		FluxMessageChannel outputChannel
+	) {
+		return input -> Flux.from(outputChannel)
+			.doOnRequest((request) ->
+					inputChannel.subscribeTo(
+							input.map((inputMessage) ->
+									MessageBuilder.fromMessage(inputMessage)
+											.removeHeader("kafka_consumer")
+											.build())));
+	}
+
+	@Bean
+	public FluxMessageChannel inputChannel() {
+		return new FluxMessageChannel();
+	}
+
+	@Bean
+	public FluxMessageChannel outputChannel() {
+		return new FluxMessageChannel();
+	}
+
+	@Bean
+	@ServiceActivator(inputChannel = "inputChannel")
+	public AggregatorFactoryBean aggregator(
+		@Nullable CorrelationStrategy correlationStrategy,
+		@Nullable ReleaseStrategy releaseStrategy,
+		@Nullable MessageGroupProcessor messageGroupProcessor,
+		@Nullable MessageGroupStore messageStore,
+		@Qualifier("outputChannel") MessageChannel outputChannel,
+		@Nullable ComponentCustomizer<AggregatorFactoryBean> aggregatorCustomizer) {
+
+		AggregatorFactoryBean aggregator = new AggregatorFactoryBean();
+		aggregator.setExpireGroupsUponCompletion(true);
+		aggregator.setSendPartialResultOnExpiry(true);
+		aggregator.setGroupTimeoutExpression(this.properties.getGroupTimeout());
+
+		if (correlationStrategy != null) {
+			aggregator.setCorrelationStrategy(correlationStrategy);
+		}
+		if (releaseStrategy != null) {
+			aggregator.setReleaseStrategy(releaseStrategy);
+		}
+
+		MessageGroupProcessor groupProcessor = messageGroupProcessor;
+
+		if (groupProcessor == null) {
+			groupProcessor = new DefaultAggregatingMessageGroupProcessor();
+			((BeanFactoryAware) groupProcessor).setBeanFactory(this.beanFactory);
+		}
+		aggregator.setProcessorBean(groupProcessor);
+
+		if (messageStore != null) {
+			aggregator.setMessageStore(messageStore);
+		}
+		aggregator.setOutputChannel(outputChannel);
+
+		if (aggregatorCustomizer != null) {
+			aggregatorCustomizer.customize(aggregator);
+		}
+
+		return aggregator;
+	}
+
+	@Bean
+	@ConditionalOnProperty(prefix = AggregatorFunctionProperties.PREFIX, name = "correlation")
+	@ConditionalOnMissingBean
+	public CorrelationStrategy correlationStrategy() {
+		return new ExpressionEvaluatingCorrelationStrategy(this.properties.getCorrelation());
+	}
+
+	@Bean
+	@ConditionalOnProperty(prefix = AggregatorFunctionProperties.PREFIX, name = "release")
+	@ConditionalOnMissingBean
+	public ReleaseStrategy releaseStrategy() {
+		return new ExpressionEvaluatingReleaseStrategy(this.properties.getRelease());
+	}
+
+	@Bean
+	@ConditionalOnProperty(prefix = AggregatorFunctionProperties.PREFIX, name = "aggregation")
+	@ConditionalOnMissingBean
+	public MessageGroupProcessor messageGroupProcessor() {
+		return new ExpressionEvaluatingMessageGroupProcessor(this.properties.getAggregation().getExpressionString());
+	}
+
+
+	@Configuration
+	@ConditionalOnMissingBean(MessageGroupStore.class)
+	@Import({
+		MessageStoreConfiguration.Mongo.class,
+		MessageStoreConfiguration.Redis.class,
+		MessageStoreConfiguration.Jdbc.class
+	})
+	protected static class MessageStoreAutoConfiguration {
+
+	}
+
+}

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

@@ -0,0 +1,122 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.aggregator;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.expression.Expression;
+
+/**
+ * Configuration properties for the Aggregator function.
+ *
+ * @author Artem Bilan
+ */
+@ConfigurationProperties("aggregator")
+public class AggregatorFunctionProperties {
+
+	static final String PREFIX = "aggregator";
+
+	/**
+	 * SpEL expression for correlation key. Default to correlationId header.
+	 */
+	private Expression correlation;
+
+	/**
+	 * SpEL expression for release strategy. Default is based on the sequenceSize header.
+	 */
+	private Expression release;
+
+	/**
+	 * SpEL expression for aggregation strategy. Default is collection of payloads.
+	 */
+	private Expression aggregation;
+
+	/**
+	 * SpEL expression for timeout to expiring uncompleted groups.
+	 */
+	private Expression groupTimeout;
+
+	/**
+	 * Message store type.
+	 */
+	private String messageStoreType = MessageStoreType.SIMPLE;
+
+	/**
+	 * Persistence message store entity: table prefix in RDBMS, collection name in MongoDb, etc.
+	 */
+	private String messageStoreEntity;
+
+	public Expression getCorrelation() {
+		return this.correlation;
+	}
+
+	public void setCorrelation(Expression correlation) {
+		this.correlation = correlation;
+	}
+
+	public Expression getRelease() {
+		return this.release;
+	}
+
+	public void setRelease(Expression release) {
+		this.release = release;
+	}
+
+	public Expression getAggregation() {
+		return this.aggregation;
+	}
+
+	public void setAggregation(Expression aggregation) {
+		this.aggregation = aggregation;
+	}
+
+	public Expression getGroupTimeout() {
+		return this.groupTimeout;
+	}
+
+	public void setGroupTimeout(Expression groupTimeout) {
+		this.groupTimeout = groupTimeout;
+	}
+
+	public String getMessageStoreEntity() {
+		return this.messageStoreEntity;
+	}
+
+	public void setMessageStoreEntity(String messageStoreEntity) {
+		this.messageStoreEntity = messageStoreEntity;
+	}
+
+	public String getMessageStoreType() {
+		return this.messageStoreType;
+	}
+
+	public void setMessageStoreType(String messageStoreType) {
+		this.messageStoreType = messageStoreType;
+	}
+
+	static final class MessageStoreType {
+
+		static final String SIMPLE = "simple";
+
+		static final String JDBC = "jdbc";
+
+		static final String MONGODB = "mongodb";
+
+		static final String REDIS = "redis";
+
+	}
+
+}

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

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2020-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.aggregator;
+
+import java.util.Properties;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.data.mongo.MongoDataAutoConfiguration;
+import org.springframework.boot.autoconfigure.data.mongo.MongoRepositoriesAutoConfiguration;
+import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
+import org.springframework.boot.autoconfigure.data.redis.RedisRepositoriesAutoConfiguration;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceTransactionManagerAutoConfiguration;
+import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
+import org.springframework.boot.env.EnvironmentPostProcessor;
+import org.springframework.core.env.ConfigurableEnvironment;
+import org.springframework.core.env.MutablePropertySources;
+import org.springframework.core.env.PropertiesPropertySource;
+
+/**
+ * An {@link EnvironmentPostProcessor} to add {@code spring.autoconfigure.exclude} property
+ * since we can't use {@code application.properties} from the library perspective.
+ *
+ * @author Artem Bilan
+ * @author Corneil du Plessis
+ */
+public class ExcludeStoresAutoConfigurationEnvironmentPostProcessor implements EnvironmentPostProcessor {
+
+	@Override
+	public void postProcessEnvironment(ConfigurableEnvironment environment, SpringApplication application) {
+		MutablePropertySources propertySources = environment.getPropertySources();
+		Properties properties = new Properties();
+
+		properties.setProperty("spring.autoconfigure.exclude",
+				DataSourceAutoConfiguration.class.getName() + ", " +
+						DataSourceTransactionManagerAutoConfiguration.class.getName() + ", " +
+						MongoAutoConfiguration.class.getName() + ", " +
+						MongoDataAutoConfiguration.class.getName() + ", " +
+						MongoRepositoriesAutoConfiguration.class.getName() + ", " +
+						RedisAutoConfiguration.class.getName() + ", " +
+						RedisRepositoriesAutoConfiguration.class.getName());
+
+		propertySources.addLast(new PropertiesPropertySource("aggregator.exclude.stores.auto-configuration", properties));
+	}
+
+}

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

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.aggregator;
+
+import java.util.Arrays;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.data.mongo.MongoDataAutoConfiguration;
+import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceTransactionManagerAutoConfiguration;
+import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Primary;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.convert.MongoCustomConversions;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.integration.jdbc.store.JdbcMessageStore;
+import org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageStore;
+import org.springframework.integration.mongodb.support.BinaryToMessageConverter;
+import org.springframework.integration.mongodb.support.MessageToBinaryConverter;
+import org.springframework.integration.redis.store.RedisMessageStore;
+import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.util.StringUtils;
+
+
+/**
+ * A helper class containing configuration classes for particular technologies
+ * to expose an appropriate {@link org.springframework.integration.store.MessageStore} bean
+ * via matched configuration properties.
+ *
+ * @author Artem Bilan
+ * @author Corneil du Plessis
+ */
+class MessageStoreConfiguration {
+
+	@ConditionalOnClass(ConfigurableMongoDbMessageStore.class)
+	@ConditionalOnProperty(prefix = AggregatorFunctionProperties.PREFIX,
+			name = "message-store-type",
+			havingValue = AggregatorFunctionProperties.MessageStoreType.MONGODB)
+	@Import({ MongoAutoConfiguration.class,
+			MongoDataAutoConfiguration.class })
+	static class Mongo {
+
+		@Bean
+		public MessageGroupStore messageStore(MongoTemplate mongoTemplate, AggregatorFunctionProperties properties) {
+			if (StringUtils.hasText(properties.getMessageStoreEntity())) {
+				return new ConfigurableMongoDbMessageStore(mongoTemplate, properties.getMessageStoreEntity());
+			}
+			else {
+				return new ConfigurableMongoDbMessageStore(mongoTemplate);
+			}
+		}
+
+		@Bean
+		@Primary
+		public MongoCustomConversions mongoDbCustomConversions() {
+			return new MongoCustomConversions(Arrays.asList(
+					new MessageToBinaryConverter(), new BinaryToMessageConverter()));
+		}
+
+	}
+
+	@ConditionalOnClass(RedisMessageStore.class)
+	@ConditionalOnProperty(prefix = AggregatorFunctionProperties.PREFIX,
+			name = "message-store-type",
+			havingValue = AggregatorFunctionProperties.MessageStoreType.REDIS)
+	@Import(RedisAutoConfiguration.class)
+	static class Redis {
+
+		@Bean
+		public MessageGroupStore messageStore(RedisTemplate<?, ?> redisTemplate) {
+			return new RedisMessageStore(redisTemplate.getConnectionFactory());
+		}
+
+	}
+
+	@ConditionalOnClass(JdbcMessageStore.class)
+	@ConditionalOnProperty(prefix = AggregatorFunctionProperties.PREFIX,
+			name = "message-store-type",
+			havingValue = AggregatorFunctionProperties.MessageStoreType.JDBC)
+	@Import({
+			DataSourceAutoConfiguration.class,
+			DataSourceTransactionManagerAutoConfiguration.class })
+	static class Jdbc {
+
+		@Bean
+		public MessageGroupStore messageStore(JdbcTemplate jdbcTemplate, AggregatorFunctionProperties properties) {
+			JdbcMessageStore messageStore = new JdbcMessageStore(jdbcTemplate);
+			if (StringUtils.hasText(properties.getMessageStoreEntity())) {
+				messageStore.setTablePrefix(properties.getMessageStoreEntity());
+			}
+			return messageStore;
+		}
+
+	}
+
+}

function/spring-aggregator-function/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,2 @@
+org.springframework.boot.env.EnvironmentPostProcessor=\
+  org.springframework.cloud.fn.aggregator.ExcludeStoresAutoConfigurationEnvironmentPostProcessor

function/spring-aggregator-function/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1 @@
+org.springframework.cloud.fn.aggregator.AggregatorFunctionConfiguration

function/spring-aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/AbstractAggregatorFunctionTests.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2020-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.aggregator;
+
+import java.util.function.Function;
+
+import reactor.core.publisher.Flux;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.integration.aggregator.AggregatingMessageHandler;
+import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.messaging.Message;
+import org.springframework.test.annotation.DirtiesContext;
+
+/**
+ * @author Artem Bilan
+ * @author Corneil du Plessis
+ */
+@SpringBootTest
+@DirtiesContext
+public abstract class AbstractAggregatorFunctionTests {
+
+	@Autowired
+	protected Function<Flux<Message<?>>, Flux<Message<?>>> aggregatorFunction;
+
+	@Autowired(required = false)
+	protected MessageGroupStore messageGroupStore;
+
+	@Autowired
+	protected AggregatingMessageHandler aggregatingMessageHandler;
+
+	@SpringBootApplication
+	public static class AggregatorFunctionTestApplication {
+		public static void main(String[] args) {
+			SpringApplication.run(AggregatorFunctionTestApplication.class, args);
+		}
+	}
+
+}

function/spring-aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/CustomPropsAndMongoMessageStoreAggregatorTests.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2020-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.aggregator;
+
+import java.time.Duration;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import org.springframework.boot.test.autoconfigure.data.mongo.AutoConfigureDataMongo;
+import org.springframework.cloud.fn.consumer.mongo.MongoDbTestContainerSupport;
+import org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageStore;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Artem Bilan
+ */
+@TestPropertySource(properties = {
+		"aggregator.correlation=T(Thread).currentThread().id",
+		"aggregator.release=!messages.?[payload == 'bar'].empty",
+		"aggregator.aggregation=#this.?[payload == 'foo'].![payload]",
+		"aggregator.messageStoreType=mongodb",
+		"aggregator.message-store-entity=aggregatorTest"
+})
+@AutoConfigureDataMongo
+public class CustomPropsAndMongoMessageStoreAggregatorTests extends AbstractAggregatorFunctionTests
+		implements MongoDbTestContainerSupport {
+
+	@DynamicPropertySource
+	static void mongoDbProperties(DynamicPropertyRegistry registry) {
+		registry.add("spring.data.mongodb.port", MONGO_CONTAINER::getFirstMappedPort);
+		registry.add("spring.data.mongodb.database", () -> "test");
+	}
+
+	@Test
+	public void test() {
+		Flux<Message<?>> input =
+				Flux.just("foo", "bar")
+						.map(GenericMessage::new);
+
+		Flux<Message<?>> output = this.aggregatorFunction.apply(input);
+
+		output.as(StepVerifier::create)
+				.assertNext((message) ->
+						assertThat(message)
+								.extracting(Message::getPayload)
+								.isInstanceOf(List.class)
+								.asList()
+								.hasSize(1)
+								.element(0).isEqualTo("foo"))
+				.thenCancel()
+				.verify(Duration.ofSeconds(10));
+
+		assertThat(this.messageGroupStore).isInstanceOf(ConfigurableMongoDbMessageStore.class);
+		assertThat(TestUtils.getPropertyValue(this.messageGroupStore, "collectionName")).isEqualTo("aggregatorTest");
+		assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore);
+	}
+
+}

function/spring-aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/DefaultAggregatorTests.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2020-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.aggregator;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.store.SimpleMessageStore;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Artem Bilan
+ * @author Corneil du Plessis
+ */
+@Disabled("Fails on CI sporadically")
+@TestPropertySource(properties = "aggregator.message-store-type=simple")
+public class DefaultAggregatorTests extends AbstractAggregatorFunctionTests {
+	private static final Logger logger = LoggerFactory.getLogger(DefaultAggregatorTests.class);
+
+	@Test
+	public void test() {
+		Flux<Message<?>> input =
+			Flux.just(MessageBuilder.withPayload("2")
+					.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation")
+					.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 2)
+					.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 2)
+					.build(),
+				MessageBuilder.withPayload("1")
+					.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation")
+					.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 1)
+					.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 2)
+					.build());
+
+		Flux<Message<?>> output = this.aggregatorFunction.apply(input.log("DefaultAggregatorTests:input"));
+		output.log("DefaultAggregatorTests:output")
+			.as(StepVerifier::create)
+			.assertNext((message) -> {
+				assertThat(message)
+					.extracting(Message::getPayload)
+					.asList()
+					.hasSize(2)
+					.contains("1", "2");
+			})
+			.thenCancel()
+			.verify(Duration.ofSeconds(30));
+
+		assertThat(this.messageGroupStore).isNull();
+		assertThat(this.aggregatingMessageHandler.getMessageStore()).isInstanceOf(SimpleMessageStore.class);
+
+	}
+}

function/spring-aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/JdbcMessageStoreAggregatorTests.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2020-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.aggregator;
+
+import java.time.Duration;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.jdbc.store.JdbcMessageStore;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Artem Bilan
+ * @author Corneil du Plessis
+ */
+@TestPropertySource(properties = "aggregator.message-store-type=jdbc")
+public class JdbcMessageStoreAggregatorTests extends AbstractAggregatorFunctionTests {
+
+	@Test
+	public void test() {
+		Flux<Message<?>> input =
+				Flux.just(MessageBuilder.withPayload("2")
+								.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation")
+								.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 2)
+								.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 2)
+								.build(),
+						MessageBuilder.withPayload("1")
+								.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation")
+								.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 1)
+								.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 2)
+								.build());
+
+		Flux<Message<?>> output = this.aggregatorFunction.apply(input);
+
+		output.as(StepVerifier::create)
+				.assertNext((message) ->
+						assertThat(message)
+								.extracting(Message::getPayload)
+								.isInstanceOf(List.class)
+								.asList()
+								.hasSize(2)
+								.contains("1", "2"))
+				.thenCancel()
+				.verify(Duration.ofSeconds(10));
+
+		assertThat(this.messageGroupStore).isInstanceOf(JdbcMessageStore.class);
+
+		assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore);
+	}
+
+}

function/spring-aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/RedisMessageStoreAggregatorTests.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2020-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.aggregator;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.cloud.fn.consumer.redis.RedisTestContainerSupport;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.redis.store.RedisMessageStore;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Artem Bilan
+ */
+@TestPropertySource(properties = "aggregator.message-store-type=redis")
+public class RedisMessageStoreAggregatorTests extends AbstractAggregatorFunctionTests implements RedisTestContainerSupport {
+
+	@DynamicPropertySource
+	static void redisProperties(DynamicPropertyRegistry registry) {
+		registry.add("spring.data.redis.url", RedisTestContainerSupport::getUri);
+	}
+
+	@Test
+	public void test() {
+		InputStream fakeNonSerializableKafkaConsumer = new ByteArrayInputStream(new byte[0]);
+
+		Flux<Message<?>> input =
+			Flux.just(MessageBuilder.withPayload("2")
+					.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation")
+					.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 2)
+					.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 2)
+					.setHeader("kafka_consumer", new ProxyFactory(fakeNonSerializableKafkaConsumer).getProxy())
+					.build(),
+				MessageBuilder.withPayload("1")
+					.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation")
+					.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 1)
+					.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 2)
+					.build());
+
+		Flux<Message<?>> output = this.aggregatorFunction.apply(input);
+
+		output.as(StepVerifier::create)
+			.assertNext((message) ->
+				assertThat(message)
+					.extracting(Message::getPayload)
+					.isInstanceOf(List.class)
+					.asList()
+					.hasSize(2)
+					.contains("1", "2"))
+			.thenCancel()
+			.verify(Duration.ofSeconds(10));
+
+		assertThat(this.messageGroupStore).isInstanceOf(RedisMessageStore.class);
+
+		assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore);
+	}
+
+}