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

Add Aggregator function & processor

Browse Source 
* 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
This commit is contained in:
Artem Bilan
2020-08-19 15:06:08 -04:00
committed by Soby Chacko
parent f88baf9b08
commit 4463c1afb6
15 changed files with 1037 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java Normal file
View File

@@ -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<Message<?>>, Flux<Message<?>>> 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> correlationStrategy,
ObjectProvider<ReleaseStrategy> releaseStrategy,
ObjectProvider<MessageGroupProcessor> messageGroupProcessor,
ObjectProvider<MessageGroupStore> 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 {
}
}

124
function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionProperties.java Normal file
View File

@@ -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";
}
}

64
function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/ExcludeStoresAutoConfigurationEnvironmentPostProcessor.java Normal file
View File

@@ -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));
}
}

148
function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/MessageStoreConfiguration.java Normal file
View File

@@ -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<Object, Object> 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;
}
}
}

4
function/aggregator-function/src/main/resources/META-INF/spring.factories Normal file
View File

@@ -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

52
function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/AbstractAggregatorFunctionTests.java Normal file
View File

@@ -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<Message<?>>, Flux<Message<?>>> aggregatorFunction;
@Autowired(required = false)
protected MessageGroupStore messageGroupStore;
@Autowired
protected AggregatingMessageHandler aggregatingMessageHandler;
@SpringBootApplication
static class AggregatorFunctionTestApplication {
}
}

70
function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/CustomPropsAndMongoMessageStoreAggregatorTests.java Normal file
View File

@@ -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<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();
assertThat(this.messageGroupStore).isInstanceOf(ConfigurableMongoDbMessageStore.class);
assertThat(TestUtils.getPropertyValue(this.messageGroupStore, "collectionName")).isEqualTo("aggregatorTest");
assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore);
}
}

68
function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/DefaultAggregatorTests.java Normal file
View File

@@ -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<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();
assertThat(this.messageGroupStore).isNull();
assertThat(this.aggregatingMessageHandler.getMessageStore()).isInstanceOf(SimpleMessageStore.class);
}
}

65
function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/GroupTimeOutAndGemfireMessageStoreAggregatorTests.java Normal file
View File

@@ -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<Message<?>> input =
Flux.just(MessageBuilder.withPayload("1")
.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation")
.build());
Flux<Message<?>> 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);
}
}

71
function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/JdbcMessageStoreAggregatorTests.java Normal file
View File

@@ -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<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();
assertThat(this.messageGroupStore).isInstanceOf(JdbcMessageStore.class);
assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore);
}
}

73
function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/RedisMessageStoreAggregatorTests.java Normal file
View File

@@ -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<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();
assertThat(this.messageGroupStore).isInstanceOf(RedisMessageStore.class);
assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore);
}
}