From b27897227da16680082ab5b15fd200203a7485f0 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 19 Aug 2020 15:06:08 -0400 Subject: [PATCH] Add Aggregator function & processor * Have all the required dependencies in class path which give us a `MessageGroupStore` implementation * Disable all their auto-configurations in the `ExcludeStoresAutoConfigurationEnvironmentPostProcessor` * Enable only those which are configured by the `aggregator.messageStoreType` property * Implement a fully reactive function around an aggregator making an interaction via a pair of `FluxMessageChannel` * Fix checkstyle issues * Change the app generator maven plugin to new coordinates --- .../aggregator-processor/README.adoc | 64 ++++++++ .../processor/aggregator-processor/pom.xml | 120 ++++++++++++++ ...onfiguration-metadata-whitelist.properties | 8 + ...dataflow-configuration-metadata.properties | 7 + .../aggregator/AggregatorProcessorTests.java | 74 +++++++++ applications/processor/pom.xml | 1 + functions/function-dependencies/pom.xml | 5 + .../function/aggregator-function/README.adoc | 25 +++ .../function/aggregator-function/pom.xml | 127 +++++++++++++++ .../AggregatorFunctionConfiguration.java | 140 +++++++++++++++++ .../AggregatorFunctionProperties.java | 124 +++++++++++++++ ...ConfigurationEnvironmentPostProcessor.java | 64 ++++++++ .../aggregator/MessageStoreConfiguration.java | 148 ++++++++++++++++++ .../main/resources/META-INF/spring.factories | 4 + .../AbstractAggregatorFunctionTests.java | 52 ++++++ ...psAndMongoMessageStoreAggregatorTests.java | 70 +++++++++ .../fn/aggregator/DefaultAggregatorTests.java | 68 ++++++++ ...AndGemfireMessageStoreAggregatorTests.java | 65 ++++++++ .../JdbcMessageStoreAggregatorTests.java | 71 +++++++++ .../RedisMessageStoreAggregatorTests.java | 73 +++++++++ functions/pom.xml | 1 + 21 files changed, 1311 insertions(+) create mode 100644 applications/processor/aggregator-processor/README.adoc create mode 100644 applications/processor/aggregator-processor/pom.xml create mode 100644 applications/processor/aggregator-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties create mode 100644 applications/processor/aggregator-processor/src/main/resources/META-INF/dataflow-configuration-metadata.properties create mode 100644 applications/processor/aggregator-processor/src/test/java/org/springframework/cloud/fn/aggregator/AggregatorProcessorTests.java create mode 100644 functions/function/aggregator-function/README.adoc create mode 100644 functions/function/aggregator-function/pom.xml create mode 100644 functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java create mode 100644 functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionProperties.java create mode 100644 functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/ExcludeStoresAutoConfigurationEnvironmentPostProcessor.java create mode 100644 functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/MessageStoreConfiguration.java create mode 100644 functions/function/aggregator-function/src/main/resources/META-INF/spring.factories create mode 100644 functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/AbstractAggregatorFunctionTests.java create mode 100644 functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/CustomPropsAndMongoMessageStoreAggregatorTests.java create mode 100644 functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/DefaultAggregatorTests.java create mode 100644 functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/GroupTimeOutAndGemfireMessageStoreAggregatorTests.java create mode 100644 functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/JdbcMessageStoreAggregatorTests.java create mode 100644 functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/RedisMessageStoreAggregatorTests.java diff --git a/applications/processor/aggregator-processor/README.adoc b/applications/processor/aggregator-processor/README.adoc new file mode 100644 index 00000000..999c6c1e --- /dev/null +++ b/applications/processor/aggregator-processor/README.adoc @@ -0,0 +1,64 @@ +//tag::ref-doc[] += Aggregator Processor + +Aggregator processor enables an application to aggregates incoming messages into groups and release them into an output destination. + +`java -jar aggregator-processor-kafka-.jar --aggregator.message-store-type=jdbc` + +Change kafka to rabbit if you want to run it against RabbitMQ. + +=== Payload + +== Options + +//tag::configuration-properties[] +$$aggregator.aggregation$$:: $$SpEL expression for aggregation strategy. Default is collection of payloads.$$ *($$Expression$$, default: `$$$$`)* +$$aggregator.correlation$$:: $$SpEL expression for correlation key. Default to correlationId header.$$ *($$Expression$$, default: `$$$$`)* +$$aggregator.group-timeout$$:: $$SpEL expression for timeout to expiring uncompleted groups.$$ *($$Expression$$, default: `$$$$`)* +$$aggregator.message-store-entity$$:: $$Persistence message store entity: table prefix in RDBMS, collection name in MongoDb, etc.$$ *($$String$$, default: `$$$$`)* +$$aggregator.message-store-type$$:: $$Message store type.$$ *($$String$$, default: `$$$$`)* +$$aggregator.release$$:: $$SpEL expression for release strategy. Default is based on the sequenceSize header.$$ *($$Expression$$, default: `$$$$`)* +$$spring.data.mongodb.authentication-database$$:: $$Authentication database name.$$ *($$String$$, default: `$$$$`)* +$$spring.data.mongodb.auto-index-creation$$:: $$Whether to enable auto-index creation.$$ *($$Boolean$$, default: `$$$$`)* +$$spring.data.mongodb.database$$:: $$Database name.$$ *($$String$$, default: `$$$$`)* +$$spring.data.mongodb.field-naming-strategy$$:: $$Fully qualified name of the FieldNamingStrategy to use.$$ *($$Class$$, default: `$$$$`)* +$$spring.data.mongodb.grid-fs-database$$:: $$GridFS database name.$$ *($$String$$, default: `$$$$`)* +$$spring.data.mongodb.host$$:: $$Mongo server host. Cannot be set with URI.$$ *($$String$$, default: `$$$$`)* +$$spring.data.mongodb.password$$:: $$Login password of the mongo server. Cannot be set with URI.$$ *($$Character[]$$, default: `$$$$`)* +$$spring.data.mongodb.port$$:: $$Mongo server port. Cannot be set with URI.$$ *($$Integer$$, default: `$$$$`)* +$$spring.data.mongodb.replica-set-name$$:: $$Required replica set name for the cluster. Cannot be set with URI.$$ *($$String$$, default: `$$$$`)* +$$spring.data.mongodb.uri$$:: $$Mongo database URI. Cannot be set with host, port, credentials and replica set name.$$ *($$String$$, default: `$$mongodb://localhost/test$$`)* +$$spring.data.mongodb.username$$:: $$Login user of the mongo server. Cannot be set with URI.$$ *($$String$$, default: `$$$$`)* +$$spring.data.mongodb.uuid-representation$$:: $$Representation to use when converting a UUID to a BSON binary value.$$ *($$UuidRepresentation$$, default: `$$java-legacy$$`, possible values: `UNSPECIFIED`,`STANDARD`,`C_SHARP_LEGACY`,`JAVA_LEGACY`,`PYTHON_LEGACY`)* +$$spring.datasource.continue-on-error$$:: $$Whether to stop if an error occurs while initializing the database.$$ *($$Boolean$$, default: `$$false$$`)* +$$spring.datasource.data$$:: $$Data (DML) script resource references.$$ *($$List$$, default: `$$$$`)* +$$spring.datasource.data-password$$:: $$Password of the database to execute DML scripts (if different).$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.data-username$$:: $$Username of the database to execute DML scripts (if different).$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.driver-class-name$$:: $$Fully qualified name of the JDBC driver. Auto-detected based on the URL by default.$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.generate-unique-name$$:: $$Whether to generate a random datasource name.$$ *($$Boolean$$, default: `$$true$$`)* +$$spring.datasource.initialization-mode$$:: $$Initialize the datasource with available DDL and DML scripts.$$ *($$DataSourceInitializationMode$$, default: `$$embedded$$`, possible values: `ALWAYS`,`EMBEDDED`,`NEVER`)* +$$spring.datasource.jndi-name$$:: $$JNDI location of the datasource. Class, url, username and password are ignored when set.$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.name$$:: $$Name of the datasource. Default to "testdb" when using an embedded database.$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.password$$:: $$Login password of the database.$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.platform$$:: $$Platform to use in the DDL or DML scripts (such as schema-${platform}.sql or data-${platform}.sql).$$ *($$String$$, default: `$$all$$`)* +$$spring.datasource.schema$$:: $$Schema (DDL) script resource references.$$ *($$List$$, default: `$$$$`)* +$$spring.datasource.schema-password$$:: $$Password of the database to execute DDL scripts (if different).$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.schema-username$$:: $$Username of the database to execute DDL scripts (if different).$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.separator$$:: $$Statement separator in SQL initialization scripts.$$ *($$String$$, default: `$$;$$`)* +$$spring.datasource.sql-script-encoding$$:: $$SQL scripts encoding.$$ *($$Charset$$, default: `$$$$`)* +$$spring.datasource.type$$:: $$Fully qualified name of the connection pool implementation to use. By default, it is auto-detected from the classpath.$$ *($$Class$$, default: `$$$$`)* +$$spring.datasource.url$$:: $$JDBC URL of the database.$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.username$$:: $$Login username of the database.$$ *($$String$$, default: `$$$$`)* +$$spring.mongodb.embedded.features$$:: $$Comma-separated list of features to enable. Uses the defaults of the configured version by default.$$ *($$Set$$, default: `$$[sync_delay]$$`)* +$$spring.mongodb.embedded.version$$:: $$Version of Mongo to use.$$ *($$String$$, default: `$$3.5.5$$`)* +$$spring.redis.client-name$$:: $$Client name to be set on connections with CLIENT SETNAME.$$ *($$String$$, default: `$$$$`)* +$$spring.redis.database$$:: $$Database index used by the connection factory.$$ *($$Integer$$, default: `$$0$$`)* +$$spring.redis.host$$:: $$Redis server host.$$ *($$String$$, default: `$$localhost$$`)* +$$spring.redis.password$$:: $$Login password of the redis server.$$ *($$String$$, default: `$$$$`)* +$$spring.redis.port$$:: $$Redis server port.$$ *($$Integer$$, default: `$$6379$$`)* +$$spring.redis.ssl$$:: $$Whether to enable SSL support.$$ *($$Boolean$$, default: `$$false$$`)* +$$spring.redis.timeout$$:: $$Connection timeout.$$ *($$Duration$$, default: `$$$$`)* +$$spring.redis.url$$:: $$Connection URL. Overrides host, port, and password. User is ignored. Example: redis://user:password@example.com:6379$$ *($$String$$, default: `$$$$`)* +//end::configuration-properties[] + +//end::ref-doc[] diff --git a/applications/processor/aggregator-processor/pom.xml b/applications/processor/aggregator-processor/pom.xml new file mode 100644 index 00000000..ea307344 --- /dev/null +++ b/applications/processor/aggregator-processor/pom.xml @@ -0,0 +1,120 @@ + + + 4.0.0 + aggregator-processor + aggregator-processor + aggregator processor apps + 3.0.0-SNAPSHOT + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + org.springframework.cloud.fn + aggregator-function + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud + spring-cloud-dataflow-apps-generator-plugin + + + aggregator + processor + ${project.version} + org.springframework.cloud.fn.aggregator.AggregatorFunctionConfiguration.class + aggregatorFunction + + + + org.springframework.cloud.fn + aggregator-function + + + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + false + + spring-releases + Spring Releases + https://repo.spring.io/release + + + + false + + spring-libs-release + Spring Libs Release + https://repo.spring.io/libs-release + + + + false + + spring-milestone-release + Spring Milestone Release + https://repo.spring.io/libs-milestone + + + + + spring-releases + Spring Releases + https://repo.spring.io/libs-release + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + diff --git a/applications/processor/aggregator-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/processor/aggregator-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..0a125844 --- /dev/null +++ b/applications/processor/aggregator-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,8 @@ +configuration-properties.classes=org.springframework.cloud.fn.aggregator.AggregatorFunctionProperties,\ + org.springframework.boot.autoconfigure.mongo.MongoProperties,\ + org.springframework.boot.autoconfigure.mongo.embedded.EmbeddedMongoProperties,\ + org.springframework.boot.autoconfigure.data.redis.RedisProperties,\ + org.springframework.boot.autoconfigure.jdbc.DataSourceProperties,\ + org.springframework.geode.boot.autoconfigure.configuration.GemFireProperties + + diff --git a/applications/processor/aggregator-processor/src/main/resources/META-INF/dataflow-configuration-metadata.properties b/applications/processor/aggregator-processor/src/main/resources/META-INF/dataflow-configuration-metadata.properties new file mode 100644 index 00000000..02e64b4c --- /dev/null +++ b/applications/processor/aggregator-processor/src/main/resources/META-INF/dataflow-configuration-metadata.properties @@ -0,0 +1,7 @@ +configuration-properties.classes=org.springframework.cloud.fn.aggregator.AggregatorFunctionProperties,\ + org.springframework.boot.autoconfigure.mongo.MongoProperties,\ + org.springframework.boot.autoconfigure.mongo.embedded.EmbeddedMongoProperties,\ + org.springframework.boot.autoconfigure.data.redis.RedisProperties,\ + org.springframework.boot.autoconfigure.jdbc.DataSourceProperties,\ + org.springframework.geode.boot.autoconfigure.configuration.GemFireProperties + diff --git a/applications/processor/aggregator-processor/src/test/java/org/springframework/cloud/fn/aggregator/AggregatorProcessorTests.java b/applications/processor/aggregator-processor/src/test/java/org/springframework/cloud/fn/aggregator/AggregatorProcessorTests.java new file mode 100644 index 00000000..cfffb751 --- /dev/null +++ b/applications/processor/aggregator-processor/src/test/java/org/springframework/cloud/fn/aggregator/AggregatorProcessorTests.java @@ -0,0 +1,74 @@ +/* + * 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.junit.jupiter.api.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.binder.test.OutputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; + +import static org.assertj.core.api.Assertions.assertThat; + +public class AggregatorProcessorTests { + + @Test + public void testFilterProcessor() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(AggregatorProcessorTestApplication.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=aggregatorFunction", + "--aggregator.message-store-type=jdbc")) { + + InputDestination processorInput = context.getBean(InputDestination.class); + OutputDestination processorOutput = context.getBean(OutputDestination.class); + + processorInput.send( + MessageBuilder.withPayload("2") + .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation") + .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 2) + .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 2) + .build()); + processorInput.send( + MessageBuilder.withPayload("1") + .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation") + .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 1) + .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 2) + .build()); + + Message receive = processorOutput.receive(10_000); + + assertThat(receive).isNotNull() + .extracting(Message::getPayload) + .extracting(String::new) + .isEqualTo("[\"2\",\"1\"]"); + } + } + + @SpringBootApplication + public static class AggregatorProcessorTestApplication { + + } + +} diff --git a/applications/processor/pom.xml b/applications/processor/pom.xml index bde1c4ab..ff657f53 100644 --- a/applications/processor/pom.xml +++ b/applications/processor/pom.xml @@ -10,6 +10,7 @@ pom + aggregator-processor bridge-processor filter-processor groovy-processor diff --git a/functions/function-dependencies/pom.xml b/functions/function-dependencies/pom.xml index cceac754..232cfe6c 100644 --- a/functions/function-dependencies/pom.xml +++ b/functions/function-dependencies/pom.xml @@ -185,6 +185,11 @@ websocket-consumer ${project.version} + + org.springframework.cloud.fn + aggregator-function + ${project.version} + org.springframework.cloud.fn filter-function diff --git a/functions/function/aggregator-function/README.adoc b/functions/function/aggregator-function/README.adoc new file mode 100644 index 00000000..2af08ebf --- /dev/null +++ b/functions/function/aggregator-function/README.adoc @@ -0,0 +1,25 @@ +# Aggregator Function + +This module provides an aggregation function that can be reused and composed in other applications. + +## Beans for injection + +You can import the `AggregatorFunctionConfiguration` in a Spring Boot application and then inject the following bean. + +`aggregatorFunction` + +You can use `aggregatorFunction` as a qualifier when injecting. + +Once injected, you can use the `apply` method of the `Function` to invoke it and get the result. + +## Configuration Options + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionProperties.java[AggregatorFunctionProperties.java] + +## Tests + +See this link:src/test/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionApplicationTests.java[test suite] for examples of how this function is used. + +## Other usage + +See this link:../../../applications/processor/aggregator-processor/README.adoc[README] where this function is used to create a Spring Cloud Stream application. diff --git a/functions/function/aggregator-function/pom.xml b/functions/function/aggregator-function/pom.xml new file mode 100644 index 00000000..2577d59f --- /dev/null +++ b/functions/function/aggregator-function/pom.xml @@ -0,0 +1,127 @@ + + + 4.0.0 + aggregator-function + 1.0.0-SNAPSHOT + aggregator-function + Spring Native Function for Aggregator + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.cloud.fn + config-common + ${project.version} + + + org.springframework.boot + spring-boot-starter-integration + + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + + + org.springframework.integration + spring-integration-mongodb + + + org.springframework.boot + spring-boot-starter-data-mongodb + runtime + + + de.flapdoodle.embed + de.flapdoodle.embed.mongo + test + + + + + org.springframework.integration + spring-integration-redis + + + org.springframework.boot + spring-boot-starter-data-redis + runtime + + + + + org.springframework.integration + spring-integration-gemfire + + + org.springframework.geode + spring-geode-starter + 1.3.2.RELEASE + + + + + org.springframework.integration + spring-integration-jdbc + + + org.springframework.boot + spring-boot-starter-jdbc + runtime + + + org.hsqldb + hsqldb + runtime + + + com.h2database + h2 + runtime + + + org.mariadb.jdbc + mariadb-java-client + runtime + + + org.postgresql + postgresql + runtime + + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + io.projectreactor + reactor-test + test + + + org.springframework.integration + spring-integration-test + test + + + + diff --git a/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java b/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java new file mode 100644 index 00000000..166693de --- /dev/null +++ b/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java @@ -0,0 +1,140 @@ +/* + * 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.function.Function; + +import reactor.core.publisher.Flux; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +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.messaging.Message; +import org.springframework.messaging.MessageChannel; + +/** + * @author Artem Bilan + */ +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties(AggregatorFunctionProperties.class) +public class AggregatorFunctionConfiguration { + + @Autowired + private AggregatorFunctionProperties properties; + + @Autowired + private BeanFactory beanFactory; + + @Bean + public Function>, Flux>> aggregatorFunction(FluxMessageChannel inputChannel, + FluxMessageChannel outputChannel) { + + return input -> Flux.from(outputChannel) + .doOnSubscribe((sub) -> inputChannel.subscribeTo(input)); + } + + @Bean + public FluxMessageChannel inputChannel() { + return new FluxMessageChannel(); + } + + @Bean + public FluxMessageChannel outputChannel() { + return new FluxMessageChannel(); + } + + @Bean + @ServiceActivator(inputChannel = "inputChannel") + public AggregatorFactoryBean aggregator( + ObjectProvider correlationStrategy, + ObjectProvider releaseStrategy, + ObjectProvider messageGroupProcessor, + ObjectProvider messageStore, + @Qualifier("outputChannel") MessageChannel outputChannel) { + + AggregatorFactoryBean aggregator = new AggregatorFactoryBean(); + aggregator.setExpireGroupsUponCompletion(true); + aggregator.setSendPartialResultOnExpiry(true); + aggregator.setGroupTimeoutExpression(this.properties.getGroupTimeout()); + + aggregator.setCorrelationStrategy(correlationStrategy.getIfAvailable()); + aggregator.setReleaseStrategy(releaseStrategy.getIfAvailable()); + + MessageGroupProcessor groupProcessor = messageGroupProcessor.getIfAvailable(); + + if (groupProcessor == null) { + groupProcessor = new DefaultAggregatingMessageGroupProcessor(); + ((BeanFactoryAware) groupProcessor).setBeanFactory(this.beanFactory); + } + aggregator.setProcessorBean(groupProcessor); + + aggregator.setMessageStore(messageStore.getIfAvailable()); + aggregator.setOutputChannel(outputChannel); + + 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.Gemfire.class, MessageStoreConfiguration.Jdbc.class }) + protected static class MessageStoreAutoConfiguration { + + } + +} diff --git a/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionProperties.java b/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionProperties.java new file mode 100644 index 00000000..05b4003c --- /dev/null +++ b/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionProperties.java @@ -0,0 +1,124 @@ +/* + * 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"; + + static final String GEMFIRE = "gemfire"; + + } + +} diff --git a/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/ExcludeStoresAutoConfigurationEnvironmentPostProcessor.java b/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/ExcludeStoresAutoConfigurationEnvironmentPostProcessor.java new file mode 100644 index 00000000..9e2fd4dc --- /dev/null +++ b/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/ExcludeStoresAutoConfigurationEnvironmentPostProcessor.java @@ -0,0 +1,64 @@ +/* + * 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.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.autoconfigure.mongo.embedded.EmbeddedMongoAutoConfiguration; +import org.springframework.boot.env.EnvironmentPostProcessor; +import org.springframework.core.env.ConfigurableEnvironment; +import org.springframework.core.env.MutablePropertySources; +import org.springframework.core.env.PropertiesPropertySource; +import org.springframework.geode.boot.autoconfigure.ClientCacheAutoConfiguration; + +/** + * 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 + */ +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() + ", " + + EmbeddedMongoAutoConfiguration.class.getName() + ", " + + ClientCacheAutoConfiguration.class.getName() + ", " + + RedisAutoConfiguration.class.getName() + ", " + + RedisRepositoriesAutoConfiguration.class.getName()); + + propertySources.addLast( + new PropertiesPropertySource("aggregator.exclude.stores.auto-configuration", properties)); + } + +} diff --git a/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/MessageStoreConfiguration.java b/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/MessageStoreConfiguration.java new file mode 100644 index 00000000..9ad7a43c --- /dev/null +++ b/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/MessageStoreConfiguration.java @@ -0,0 +1,148 @@ +/* + * 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.apache.geode.cache.GemFireCache; +import org.apache.geode.cache.Region; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +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.boot.autoconfigure.mongo.embedded.EmbeddedMongoAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.Primary; +import org.springframework.data.gemfire.client.ClientRegionFactoryBean; +import org.springframework.data.gemfire.config.annotation.EnablePdx; +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.geode.boot.autoconfigure.ClientCacheAutoConfiguration; +import org.springframework.integration.gemfire.store.GemfireMessageStore; +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 + */ +class MessageStoreConfiguration { + + @ConditionalOnClass(ConfigurableMongoDbMessageStore.class) + @ConditionalOnProperty(prefix = AggregatorFunctionProperties.PREFIX, + name = "message-store-type", + havingValue = AggregatorFunctionProperties.MessageStoreType.MONGODB) + @Import({ MongoAutoConfiguration.class, + MongoDataAutoConfiguration.class, + EmbeddedMongoAutoConfiguration.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(GemfireMessageStore.class) + @ConditionalOnProperty(prefix = AggregatorFunctionProperties.PREFIX, + name = "message-store-type", + havingValue = AggregatorFunctionProperties.MessageStoreType.GEMFIRE) + @Import(ClientCacheAutoConfiguration.class) + @EnablePdx + static class Gemfire { + + @Bean + @ConditionalOnMissingBean + public ClientRegionFactoryBean gemfireRegion(GemFireCache cache, AggregatorFunctionProperties properties) { + ClientRegionFactoryBean clientRegionFactoryBean = new ClientRegionFactoryBean<>(); + clientRegionFactoryBean.setCache(cache); + clientRegionFactoryBean.setName(properties.getMessageStoreEntity()); + return clientRegionFactoryBean; + } + + @Bean + public MessageGroupStore messageStore(Region region) { + return new GemfireMessageStore(region); + } + + } + + @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; + } + + } + +} diff --git a/functions/function/aggregator-function/src/main/resources/META-INF/spring.factories b/functions/function/aggregator-function/src/main/resources/META-INF/spring.factories new file mode 100644 index 00000000..55a57ec9 --- /dev/null +++ b/functions/function/aggregator-function/src/main/resources/META-INF/spring.factories @@ -0,0 +1,4 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ + org.springframework.cloud.fn.aggregator.AggregatorFunctionConfiguration +org.springframework.boot.env.EnvironmentPostProcessor=\ + org.springframework.cloud.fn.aggregator.ExcludeStoresAutoConfigurationEnvironmentPostProcessor diff --git a/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/AbstractAggregatorFunctionTests.java b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/AbstractAggregatorFunctionTests.java new file mode 100644 index 00000000..2408abe9 --- /dev/null +++ b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/AbstractAggregatorFunctionTests.java @@ -0,0 +1,52 @@ +/* + * 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.function.Function; + +import reactor.core.publisher.Flux; + +import org.springframework.beans.factory.annotation.Autowired; +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 + */ +@SpringBootTest +@DirtiesContext +public abstract class AbstractAggregatorFunctionTests { + + @Autowired + protected Function>, Flux>> aggregatorFunction; + + @Autowired(required = false) + protected MessageGroupStore messageGroupStore; + + @Autowired + protected AggregatingMessageHandler aggregatingMessageHandler; + + @SpringBootApplication + static class AggregatorFunctionTestApplication { + + } + +} diff --git a/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/CustomPropsAndMongoMessageStoreAggregatorTests.java b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/CustomPropsAndMongoMessageStoreAggregatorTests.java new file mode 100644 index 00000000..da961520 --- /dev/null +++ b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/CustomPropsAndMongoMessageStoreAggregatorTests.java @@ -0,0 +1,70 @@ +/* + * 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.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.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.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 { + + @Test + public void test() { + Flux> input = + Flux.just("foo", "bar") + .map(GenericMessage::new); + + Flux> 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(); + + assertThat(this.messageGroupStore).isInstanceOf(ConfigurableMongoDbMessageStore.class); + assertThat(TestUtils.getPropertyValue(this.messageGroupStore, "collectionName")).isEqualTo("aggregatorTest"); + assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore); + } + +} diff --git a/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/DefaultAggregatorTests.java b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/DefaultAggregatorTests.java new file mode 100644 index 00000000..62cd1f7f --- /dev/null +++ b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/DefaultAggregatorTests.java @@ -0,0 +1,68 @@ +/* + * 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.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.store.SimpleMessageStore; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Artem Bilan + */ +public class DefaultAggregatorTests extends AbstractAggregatorFunctionTests { + + @Test + public void test() { + Flux> 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> 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(); + + assertThat(this.messageGroupStore).isNull(); + assertThat(this.aggregatingMessageHandler.getMessageStore()).isInstanceOf(SimpleMessageStore.class); + } + +} diff --git a/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/GroupTimeOutAndGemfireMessageStoreAggregatorTests.java b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/GroupTimeOutAndGemfireMessageStoreAggregatorTests.java new file mode 100644 index 00000000..abdbfd61 --- /dev/null +++ b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/GroupTimeOutAndGemfireMessageStoreAggregatorTests.java @@ -0,0 +1,65 @@ +/* + * 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.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.gemfire.store.GemfireMessageStore; +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 + */ +@TestPropertySource(properties = { + "aggregator.message-store-type=gemfire", + "aggregator.groupTimeout=10" }) +public class GroupTimeOutAndGemfireMessageStoreAggregatorTests extends AbstractAggregatorFunctionTests { + + @Test + public void test() { + Flux> input = + Flux.just(MessageBuilder.withPayload("1") + .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation") + .build()); + + Flux> output = this.aggregatorFunction.apply(input); + + output.as(StepVerifier::create) + .assertNext((message) -> + assertThat(message) + .extracting(Message::getPayload) + .isInstanceOf(List.class) + .asList() + .hasSize(1) + .contains("1")) + .thenCancel() + .verify(); + + assertThat(this.messageGroupStore).isInstanceOf(GemfireMessageStore.class); + assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore); + } + +} diff --git a/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/JdbcMessageStoreAggregatorTests.java b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/JdbcMessageStoreAggregatorTests.java new file mode 100644 index 00000000..6357489c --- /dev/null +++ b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/JdbcMessageStoreAggregatorTests.java @@ -0,0 +1,71 @@ +/* + * 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.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 + */ +@TestPropertySource(properties = "aggregator.message-store-type=jdbc") +public class JdbcMessageStoreAggregatorTests extends AbstractAggregatorFunctionTests { + + @Test + public void test() { + Flux> 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> 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(); + + assertThat(this.messageGroupStore).isInstanceOf(JdbcMessageStore.class); + + assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore); + } + +} diff --git a/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/RedisMessageStoreAggregatorTests.java b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/RedisMessageStoreAggregatorTests.java new file mode 100644 index 00000000..e99a9fa6 --- /dev/null +++ b/functions/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/RedisMessageStoreAggregatorTests.java @@ -0,0 +1,73 @@ +/* + * 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.List; + +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +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.TestPropertySource; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Artem Bilan + */ +@TestPropertySource(properties = "aggregator.message-store-type=redis") +@Disabled("Needs real Redis Server to be run") +public class RedisMessageStoreAggregatorTests extends AbstractAggregatorFunctionTests { + + @Test + public void test() { + Flux> 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> 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(); + + assertThat(this.messageGroupStore).isInstanceOf(RedisMessageStore.class); + + assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore); + } + +} diff --git a/functions/pom.xml b/functions/pom.xml index a90e612a..c4271ab3 100644 --- a/functions/pom.xml +++ b/functions/pom.xml @@ -72,6 +72,7 @@ consumer/twitter-consumer consumer/wavefront-consumer + function/aggregator-function function/filter-function function/header-enricher-function function/http-request-function