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

Make Mongo & JDBC modules as auto-config

Browse Source 
* Also make a `spring-splitter-function` as an auto-config since it used in Mongo & JDBC suppliers
* Fix all their Checkstyle violations
This commit is contained in:
Artem Bilan
2024-01-09 13:10:18 -05:00
parent 26479fb1bc
commit 8ef105b9c5
37 changed files with 184 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
consumer/spring-jdbc-consumer/README.adoc
View File

@@ -5,7 +5,7 @@ The consumer uses the `JdbcMessageHandler` from Spring Integration.
## Beans for injection
You can import `JdbcConsumerConfiguration` in the application and then inject the following bean.
The `JdbcConsumerConfiguration` auto-configuration provides the following bean:
`Consumer<Message<?>> jdbcConsumer`

93
consumer/spring-jdbc-consumer/src/main/java/org/springframework/cloud/fn/consumer/jdbc/JdbcConsumerConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2023 the original author or authors.
* Copyright 2020-2024 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.
@@ -29,30 +29,28 @@ import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ResourceLoader;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.EvaluationException;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.SpelParseException;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor;
import org.springframework.integration.aggregator.MessageCountReleaseStrategy;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.config.AggregatorFactoryBean;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.jdbc.JdbcMessageHandler;
import org.springframework.integration.jdbc.SqlParameterSourceFactory;
import org.springframework.integration.json.JsonPropertyAccessor;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.MutableMessage;
@@ -68,6 +66,8 @@ import org.springframework.util.MimeTypeUtils;
import org.springframework.util.MultiValueMap;
/**
* Auto-configuration for JDBC consumer.
*
* @author Eric Bottard
* @author Thomas Risberg
* @author Robert St. John
@@ -76,25 +76,20 @@ import org.springframework.util.MultiValueMap;
* @author Soby Chacko
* @author Szabolcs Stremler
*/
@Configuration
@AutoConfiguration(after = DataSourceAutoConfiguration.class)
@EnableConfigurationProperties(JdbcConsumerProperties.class)
public class JdbcConsumerConfiguration {
private static final Log logger = LogFactory.getLog(JdbcConsumerConfiguration.class);
private static final Log LOGGER = LogFactory.getLog(JdbcConsumerConfiguration.class);
private static final Object NOT_SET = new Object();
private static final SpelExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
private final JdbcConsumerProperties properties;
private SpelExpressionParser spelExpressionParser = new SpelExpressionParser();
private EvaluationContext evaluationContext;
public JdbcConsumerConfiguration(JdbcConsumerProperties properties, BeanFactory beanFactory) {
public JdbcConsumerConfiguration(JdbcConsumerProperties properties) {
this.properties = properties;
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(beanFactory);
StandardEvaluationContext standardEvaluationContext = (StandardEvaluationContext) this.evaluationContext;
standardEvaluationContext.addPropertyAccessor(new JsonPropertyAccessor());
}
@Bean
@@ -128,18 +123,18 @@ public class JdbcConsumerConfiguration {
IntegrationFlow jdbcConsumerFlow(@Qualifier("aggregator") MessageHandler aggregator,
JdbcMessageHandler jdbcMessageHandler) {
final IntegrationFlowBuilder builder = IntegrationFlow.from(Consumer.class,
gateway -> gateway.beanName("jdbcConsumer"));
if (properties.getBatchSize() > 1 || properties.getIdleTimeout() > 0) {
builder.handle(aggregator);
}
return builder.handle(jdbcMessageHandler).get();
return (flow) -> {
if (this.properties.getBatchSize() > 1 || this.properties.getIdleTimeout() > 0) {
flow.handle(aggregator);
}
flow.handle(jdbcMessageHandler);
};
}
@Bean
FactoryBean<MessageHandler> aggregator(MessageGroupStore messageGroupStore) {
AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
aggregatorFactoryBean.setCorrelationStrategy(message -> message.getPayload().getClass().getName());
aggregatorFactoryBean.setCorrelationStrategy((message) -> message.getPayload().getClass().getName());
aggregatorFactoryBean.setReleaseStrategy(new MessageCountReleaseStrategy(this.properties.getBatchSize()));
if (this.properties.getIdleTimeout() >= 0) {
aggregatorFactoryBean.setGroupTimeoutExpression(new ValueExpression<>(this.properties.getIdleTimeout()));
@@ -160,19 +155,20 @@ public class JdbcConsumerConfiguration {
}
@Bean
public JdbcMessageHandler jdbcMessageHandler(DataSource dataSource) {
public JdbcMessageHandler jdbcMessageHandler(DataSource dataSource,
@Qualifier(IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME) EvaluationContext evaluationContext) {
final MultiValueMap<String, Expression> columnExpressionVariations = new LinkedMultiValueMap<>();
for (Map.Entry<String, String> entry : this.properties.getColumnsMap().entrySet()) {
String value = entry.getValue();
columnExpressionVariations.add(entry.getKey(), this.spelExpressionParser.parseExpression(value));
columnExpressionVariations.add(entry.getKey(), EXPRESSION_PARSER.parseExpression(value));
if (!value.startsWith("payload")) {
String qualified = "payload." + value;
try {
columnExpressionVariations.add(entry.getKey(),
this.spelExpressionParser.parseExpression(qualified));
columnExpressionVariations.add(entry.getKey(), EXPRESSION_PARSER.parseExpression(qualified));
}
catch (SpelParseException e) {
logger.info("failed to parse qualified fallback expression " + qualified
catch (SpelParseException ex) {
LOGGER.info("failed to parse qualified fallback expression " + qualified
+ "; be sure your expression uses the 'payload.' prefix where necessary");
}
}
@@ -191,10 +187,9 @@ public class JdbcConsumerConfiguration {
if (message.getPayload() instanceof Iterable) {
Stream<Object> messageStream = StreamSupport
.stream(((Iterable<?>) message.getPayload()).spliterator(), false)
.map(payload -> {
if (payload instanceof byte[]) {
return convertibleContentType(contentType) ? new String(((byte[]) payload))
: payload;
.map((payload) -> {
if (payload instanceof byte[] bytes) {
return (convertibleContentType(contentType)) ? new String(bytes) : bytes;
}
else {
return payload;
@@ -205,7 +200,7 @@ public class JdbcConsumerConfiguration {
}
else {
if (convertibleContentType(contentType)) {
convertedMessage = new MutableMessage<>(new String(((byte[]) message.getPayload())),
convertedMessage = new MutableMessage<>(new String((byte[]) message.getPayload()),
message.getHeaders());
}
}
@@ -214,7 +209,7 @@ public class JdbcConsumerConfiguration {
}
};
SqlParameterSourceFactory parameterSourceFactory = new ParameterFactory(columnExpressionVariations,
this.evaluationContext);
evaluationContext);
jdbcMessageHandler.setSqlParameterSourceFactory(parameterSourceFactory);
return jdbcMessageHandler;
}
@@ -227,7 +222,7 @@ public class JdbcConsumerConfiguration {
ResourceDatabasePopulator databasePopulator = new ResourceDatabasePopulator();
databasePopulator.setIgnoreFailedDrops(true);
dataSourceInitializer.setDatabasePopulator(databasePopulator);
if ("true".equals(properties.getInitialize())) {
if ("true".equals(this.properties.getInitialize())) {
databasePopulator.addScript(new DefaultInitializationScriptResource(this.properties.getTableName(),
this.properties.getColumnsMap().keySet()));
}
@@ -237,23 +232,19 @@ public class JdbcConsumerConfiguration {
return dataSourceInitializer;
}
private static final class ParameterFactory implements SqlParameterSourceFactory {
@MessagingGateway(name = "jdbcConsumer", defaultRequestChannel = "jdbcConsumerFlow.input")
public interface MessageConsumer extends Consumer<Message<?>> {
private final MultiValueMap<String, Expression> columnExpressions;
}
private final EvaluationContext context;
ParameterFactory(MultiValueMap<String, Expression> columnExpressions, EvaluationContext context) {
this.columnExpressions = columnExpressions;
this.context = context;
}
private record ParameterFactory(MultiValueMap<String, Expression> columnExpressions,
EvaluationContext context) implements SqlParameterSourceFactory {
@Override
public SqlParameterSource createParameterSource(Object o) {
if (!(o instanceof Message)) {
if (!(o instanceof Message<?> message)) {
throw new IllegalArgumentException("Unable to handle type " + o.getClass().getName());
}
Message<?> message = (Message<?>) o;
MapSqlParameterSource parameterSource = new MapSqlParameterSource();
for (Map.Entry<String, List<Expression>> entry : this.columnExpressions.entrySet()) {
String key = entry.getKey();
@@ -262,16 +253,16 @@ public class JdbcConsumerConfiguration {
EvaluationException lastException = null;
for (Expression spel : spels) {
try {
value = spel.getValue(context, message);
value = spel.getValue(this.context, message);
break;
}
catch (EvaluationException e) {
lastException = e;
catch (EvaluationException ex) {
lastException = ex;
}
}
if (value == NOT_SET) {
if (lastException != null) {
logger.info("Could not find value for column '" + key + "': " + lastException.getMessage());
LOGGER.info("Could not find value for column '" + key + "': " + lastException.getMessage());
}
parameterSource.addValue(key, null);
}

4
consumer/spring-jdbc-consumer/src/main/java/org/springframework/cloud/fn/consumer/jdbc/JdbcConsumerProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2024 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.
@@ -22,6 +22,8 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* The configuration properties for JDBC consumer.
*
* @author Eric Bottard
* @author Artem Bilan
* @author Oliver Flasch

6
consumer/spring-jdbc-consumer/src/main/java/org/springframework/cloud/fn/consumer/jdbc/ShorthandMapConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2024 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.
@@ -46,7 +46,7 @@ public class ShorthandMapConverter implements Converter<String, Map<String, Stri
for (String mapping : mappings) {
// Turn backslash-comma back to comma
String unescaped = mapping.trim().replace("\\,", ",");
if (unescaped.length() == 0) {
if (unescaped.isEmpty()) {
continue;
}
// Split on colon, if not preceded by backslash
@@ -54,7 +54,7 @@ public class ShorthandMapConverter implements Converter<String, Map<String, Stri
Assert.isTrue(keyValuePair.length <= 2, "'" + unescaped
+ "' could not be parsed to a 'key:value' pair or simple 'key' with implicit value");
String key = keyValuePair[0].trim().replace("\\:", ":");
String value = keyValuePair.length == 2 ? keyValuePair[1].trim().replace("\\:", ":") : key;
String value = (keyValuePair.length == 2) ? keyValuePair[1].trim().replace("\\:", ":") : key;
result.put(key, value);
}
return result;

4
consumer/spring-jdbc-consumer/src/main/java/org/springframework/cloud/fn/consumer/jdbc/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The JDBC consumer auto-configuration support.
*/
package org.springframework.cloud.fn.consumer.jdbc;

1
consumer/spring-jdbc-consumer/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.consumer.jdbc.JdbcConsumerConfiguration

1
consumer/spring-jdbc-consumer/src/main/resources/application.properties
View File

@@ -1 +0,0 @@
spring.integration.jdbc.initialize-schema=NEVER

4
consumer/spring-jdbc-consumer/src/test/java/org/springframework/cloud/fn/consumer/jdbc/BatchInsertTimeoutTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2024 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.
@@ -45,7 +45,7 @@ public class BatchInsertTimeoutTests extends JdbcConsumerApplicationTests {
}
Awaitility.await()
.until(() -> jdbcOperations.queryForObject("select count(*) from messages", Integer.class),
value -> value == numberOfInserts);
(value) -> value == numberOfInserts);
}
}

4
consumer/spring-jdbc-consumer/src/test/java/org/springframework/cloud/fn/consumer/jdbc/JdbcConsumerApplicationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2024 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.
@@ -49,7 +49,7 @@ public class JdbcConsumerApplicationTests {
jdbcOperations.execute("DROP TABLE MESSAGES IF EXISTS");
}
static class Payload {
public static class Payload {
private String a;

6
consumer/spring-mongodb-consumer/README.adoc
View File

@@ -4,7 +4,7 @@ A consumer that allows you to insert records into MongoDB.
## Beans for injection
You can import `MongoDbConsumerConfiguration` in the application and then inject one of the following beans.
The `MongoDbConsumerConfiguration` auto-configuration provides the following beans:
`Function<Message<?>, Mono<Void>> mongodbConsumerFunction` - Allows you to subscribe.
@@ -18,13 +18,13 @@ The return value from the function can be ignored as this is used as a consumer
All configuration properties are prefixed with `mongodb.consumer`.
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/mongo/MongoDBConsumerProperties.java[MongoDBConsumerProperties].
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/mongo/MongoDbConsumerProperties.java[MongoDBConsumerProperties].
A `ComponentCustomizer<ReactiveMongoDbStoringMessageHandler>` bean can be added in the target project to provide any custom options for the `ReactiveMongoDbStoringMessageHandler` configuration used by the `mongodbConsumer`.
## Examples
See this link:src/test/java/org/springframework/cloud/fn/consumer/mongo/MongoDBConsumerApplicationTests.java[test suite] for the various ways, this consumer is used.
See this link:src/test/java/org/springframework/cloud/fn/consumer/mongo/MongoDbConsumerApplicationTests.java[test suite] for the various ways, this consumer is used.
## Other usage

11
consumer/spring-mongodb-consumer/src/main/java/org/springframework/cloud/fn/consumer/mongo/MongoDbConsumerConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2022 the original author or authors.
* Copyright 2017-2024 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.
@@ -21,10 +21,11 @@ import java.util.function.Function;
import reactor.core.publisher.Mono;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.mongo.MongoReactiveAutoConfiguration;
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.data.mongodb.core.ReactiveMongoTemplate;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
@@ -41,8 +42,8 @@ import org.springframework.messaging.ReactiveMessageHandler;
* @author David Turanski
*
*/
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties({ MongoDbConsumerProperties.class })
@AutoConfiguration(after = MongoReactiveAutoConfiguration.class)
@EnableConfigurationProperties(MongoDbConsumerProperties.class)
public class MongoDbConsumerConfiguration {
private final MongoDbConsumerProperties properties;
@@ -56,7 +57,7 @@ public class MongoDbConsumerConfiguration {
@Bean
public Consumer<Message<?>> mongodbConsumer(Function<Message<?>, Mono<Void>> mongodbConsumerFunction) {
return message -> mongodbConsumerFunction.apply(message).subscribe();
return (message) -> mongodbConsumerFunction.apply(message).subscribe();
}
@Bean

6
consumer/spring-mongodb-consumer/src/main/java/org/springframework/cloud/fn/consumer/mongo/MongoDbConsumerProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* Copyright 2019-2024 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.
@@ -24,6 +24,8 @@ import org.springframework.util.StringUtils;
import org.springframework.validation.annotation.Validated;
/**
* The configuration properties for MongoDB consumer.
*
* @author Artem Bilan
* @author David Turanski
*
@@ -55,7 +57,7 @@ public class MongoDbConsumerProperties {
}
public Expression getCollectionExpression() {
return collectionExpression;
return this.collectionExpression;
}
@AssertTrue(message = "One of 'collection' or 'collectionExpression' is required")

4
consumer/spring-mongodb-consumer/src/main/java/org/springframework/cloud/fn/consumer/mongo/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The MongoDB consumer auto-configuration support.
*/
package org.springframework.cloud.fn.consumer.mongo;

1
consumer/spring-mongodb-consumer/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.consumer.mongo.MongoDbConsumerConfiguration

32
consumer/spring-mongodb-consumer/src/test/java/org/springframework/cloud/fn/consumer/mongo/MongoDbConsumerApplicationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2024 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.
@@ -33,8 +33,7 @@ import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
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.annotation.DirtiesContext;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
@@ -43,15 +42,10 @@ import static org.awaitility.Awaitility.await;
* @author David Turanski
* @author Chris Bono
*/
@SpringBootTest(properties = { "mongodb.consumer.collection=testing" })
@SpringBootTest(properties = "mongodb.consumer.collection=testing")
@DirtiesContext
class MongoDbConsumerApplicationTests implements MongoDbTestContainerSupport {
@DynamicPropertySource
static void mongoDbProperties(DynamicPropertyRegistry registry) {
registry.add("spring.data.mongodb.port", MONGO_CONTAINER::getFirstMappedPort);
registry.add("spring.data.mongodb.database", () -> "test");
}
@Autowired
private MongoDbConsumerProperties properties;
@@ -73,28 +67,26 @@ class MongoDbConsumerApplicationTests implements MongoDbTestContainerSupport {
Flux<Message<?>> messages = Flux.just(new GenericMessage<>(data1), new GenericMessage<>(data2),
new GenericMessage<>("{\"my_data\": \"THE DATA\"}"));
messages.map(message -> {
messages.map((message) -> {
mongodbConsumer.accept(message);
return message;
}).subscribe();
await().timeout(Duration.ofSeconds(10))
.until(() -> mongoTemplate.findAll(Document.class, properties.getCollection()).count().block() == 3L);
.untilAsserted(
() -> assertThat(mongoTemplate.findAll(Document.class, properties.getCollection()).count().block())
.isEqualTo(3L));
StepVerifier
.create(this.mongoTemplate.findAll(Document.class, properties.getCollection())
.sort(Comparator.comparing(d -> d.get("_id").toString())))
.assertNext(document -> {
assertThat(document.get("foo")).isEqualTo("bar");
})
.assertNext(document -> {
.sort(Comparator.comparing((d) -> d.get("_id").toString())))
.assertNext((document) -> assertThat(document.get("foo")).isEqualTo("bar"))
.assertNext((document) -> {
assertThat(document.get("firstName")).isEqualTo("Foo");
assertThat(document.get("lastName")).isEqualTo("Bar");
})
.assertNext(document -> {
assertThat(document.get("my_data")).isEqualTo("THE DATA");
})
.assertNext((document) -> assertThat(document.get("my_data")).isEqualTo("THE DATA"))
.verifyComplete();
}

16
consumer/spring-mongodb-consumer/src/test/java/org/springframework/cloud/fn/consumer/mongo/MongoDbTestContainerSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022-2022 the original author or authors.
* Copyright 2022-2024 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.
@@ -16,12 +16,13 @@
package org.springframework.cloud.fn.consumer.mongo;
import java.time.Duration;
import org.junit.jupiter.api.BeforeAll;
import org.testcontainers.containers.MongoDBContainer;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
/**
* Provides a static {@link MongoDBContainer} that can be shared across test classes.
*
@@ -30,14 +31,19 @@ import org.testcontainers.junit.jupiter.Testcontainers;
@Testcontainers(disabledWithoutDocker = true)
public interface MongoDbTestContainerSupport {
MongoDBContainer MONGO_CONTAINER = new MongoDBContainer("mongo:6.0.6").withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3);
MongoDBContainer MONGO_CONTAINER = new MongoDBContainer("mongo");
@BeforeAll
static void startContainer() {
MONGO_CONTAINER.start();
}
@DynamicPropertySource
static void mongoDbProperties(DynamicPropertyRegistry registry) {
registry.add("spring.data.mongodb.port", MONGO_CONTAINER::getFirstMappedPort);
registry.add("spring.data.mongodb.database", () -> "test");
}
static String mongoDbUri() {
return "mongodb://localhost:" + MONGO_CONTAINER.getFirstMappedPort();
}

2
function/spring-splitter-function/README.adoc
View File

@@ -4,7 +4,7 @@ This module provides a header enricher function that can be reused and composed
## Beans for injection
You can import the `SpliiterFunctionConfiguration` in a Spring Boot application and then inject the following bean.
The `SpliiterFunctionConfiguration` auto-configuration provides the following bean:
`splitterFunction`

18
function/spring-splitter-function/src/main/java/org/springframework/cloud/fn/splitter/SplitterFunctionConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2011-2020 the original author or authors.
* Copyright 2011-2024 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.
@@ -23,14 +23,13 @@ import java.util.function.Function;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.AnyNestedCondition;
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.Conditional;
import org.springframework.context.annotation.Configuration;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
import org.springframework.integration.file.splitter.FileSplitter;
import org.springframework.integration.splitter.AbstractMessageSplitter;
@@ -39,7 +38,13 @@ import org.springframework.integration.splitter.ExpressionEvaluatingSplitter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@Configuration
/**
* Auto-configuration for Splitter function.
*
* @author Artem Bilan
* @author Soby Chacko
*/
@AutoConfiguration
@EnableConfigurationProperties(SplitterFunctionProperties.class)
public class SplitterFunctionConfiguration {
@@ -50,7 +55,7 @@ public class SplitterFunctionConfiguration {
messageSplitter.setApplySequence(splitterFunctionProperties.isApplySequence());
ThreadLocalFluxSinkMessageChannel outputChannel = new ThreadLocalFluxSinkMessageChannel();
messageSplitter.setOutputChannel(outputChannel);
return message -> {
return (message) -> {
messageSplitter.handleMessage(message);
return outputChannel.publisherThreadLocal.get();
};
@@ -59,8 +64,7 @@ public class SplitterFunctionConfiguration {
@Bean
@ConditionalOnProperty(prefix = "splitter", name = "expression")
public AbstractMessageSplitter expressionSplitter(SplitterFunctionProperties splitterFunctionProperties) {
return new ExpressionEvaluatingSplitter(
new SpelExpressionParser().parseExpression(splitterFunctionProperties.getExpression()));
return new ExpressionEvaluatingSplitter(splitterFunctionProperties.getExpression());
}
@Bean

11
function/spring-splitter-function/src/main/java/org/springframework/cloud/fn/splitter/SplitterFunctionProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* Copyright 2019-2024 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.
@@ -19,10 +19,11 @@ package org.springframework.cloud.fn.splitter;
import jakarta.validation.constraints.AssertTrue;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.expression.Expression;
import org.springframework.validation.annotation.Validated;
/**
* Configuration properties for the Splitter Processor app.
* Configuration properties for the Splitter function.
*
* @author Gary Russell
* @author Artem Bilan
@@ -34,7 +35,7 @@ public class SplitterFunctionProperties {
/**
* A SpEL expression for splitting payloads.
*/
private String expression;
private Expression expression;
/**
* When expression is null, delimiters to use when tokenizing {@link String} payloads.
@@ -63,11 +64,11 @@ public class SplitterFunctionProperties {
*/
private boolean applySequence = true;
public String getExpression() {
public Expression getExpression() {
return this.expression;
}
public void setExpression(String expression) {
public void setExpression(Expression expression) {
this.expression = expression;
}

4
function/spring-splitter-function/src/main/java/org/springframework/cloud/fn/splitter/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The Splitter function auto-configuration support.
*/
package org.springframework.cloud.fn.splitter;

1
function/spring-splitter-function/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.splitter.SplitterFunctionConfiguration

4
function/spring-splitter-function/src/test/java/org/springframework/cloud/fn/splitter/SplitterFunctionApplicationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2011-2020 the original author or authors.
* Copyright 2011-2024 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.
@@ -40,7 +40,7 @@ public class SplitterFunctionApplicationTests {
@Test
public void testExpressionSplitter() {
List<Message<?>> messageList = this.splitter.apply(new GenericMessage<>("hello,world"));
assertThat(messageList).extracting(m -> m.getPayload().toString()).contains("hello", "world");
assertThat(messageList).extracting((m) -> m.getPayload().toString()).contains("hello", "world");
}
@SpringBootApplication

2
supplier/spring-jdbc-supplier/README.adoc
View File

@@ -7,7 +7,7 @@ When you have use-cases such as periodical execution of a Database query, based
## Beans for injection
You can import the `JdbcSupplierConfiguration` in the application and then inject the following bean.
The `JdbcSupplierConfiguration` auto-configuration provides the following bean:
`jdbcSupplier`

17
supplier/spring-jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/JdbcSupplierConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2024 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.
@@ -24,26 +24,27 @@ import javax.sql.DataSource;
import reactor.core.publisher.Flux;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.common.config.ComponentCustomizer;
import org.springframework.cloud.fn.splitter.SplitterFunctionConfiguration;
import org.springframework.cloud.function.context.PollableBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.jdbc.JdbcPollingChannelAdapter;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
/**
* JDBC supplier auto-configuration.
*
* @author Soby Chacko
* @author Artem Bilan
*/
@Configuration(proxyBeanMethods = false)
@AutoConfiguration(after = { DataSourceAutoConfiguration.class, SplitterFunctionConfiguration.class })
@EnableConfigurationProperties(JdbcSupplierProperties.class)
@Import(SplitterFunctionConfiguration.class)
public class JdbcSupplierConfiguration {
private final JdbcSupplierProperties properties;
@@ -74,12 +75,12 @@ public class JdbcSupplierConfiguration {
@ConditionalOnProperty(prefix = "jdbc.supplier", name = "split", matchIfMissing = true)
public Supplier<Flux<Message<?>>> splittedSupplier(MessageSource<Object> jdbcMessageSource,
Function<Message<?>, List<Message<?>>> splitterFunction) {
return () -> {
Message<?> received = jdbcMessageSource.receive();
if (received != null) {
return Flux.fromIterable(splitterFunction.apply(received)); // multiple
// Message<Map<String,
// Object>>
// multiple Message<Map<String, Object>>
return Flux.fromIterable(splitterFunction.apply(received));
}
else {
return Flux.empty();

12
supplier/spring-jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/JdbcSupplierProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* Copyright 2019-2024 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.
@@ -22,6 +22,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.validation.annotation.Validated;
/**
* JDBC supplier configuration properties.
*
* @author Soby Chacko
* @author Artem Bilan
*/
@@ -51,7 +53,7 @@ public class JdbcSupplierProperties {
@NotNull
public String getQuery() {
return query;
return this.query;
}
public void setQuery(String query) {
@@ -59,7 +61,7 @@ public class JdbcSupplierProperties {
}
public String getUpdate() {
return update;
return this.update;
}
public void setUpdate(String update) {
@@ -67,7 +69,7 @@ public class JdbcSupplierProperties {
}
public boolean isSplit() {
return split;
return this.split;
}
public void setSplit(boolean split) {
@@ -75,7 +77,7 @@ public class JdbcSupplierProperties {
}
public int getMaxRows() {
return maxRows;
return this.maxRows;
}
public void setMaxRows(int maxRows) {

4
supplier/spring-jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The JDBC supplier auto-configuration support.
*/
package org.springframework.cloud.fn.supplier.jdbc;

1
supplier/spring-jdbc-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.supplier.jdbc.JdbcSupplierConfiguration

1
supplier/spring-jdbc-supplier/src/main/resources/application.properties
View File

@@ -1 +0,0 @@
spring.integration.jdbc.initialize-schema=NEVER

29
supplier/spring-jdbc-supplier/src/test/java/org/springframework/cloud/fn/supplier/jdbc/DefaultJdbcSupplierTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2022 the original author or authors.
* Copyright 2020-2024 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.
@@ -26,16 +26,13 @@ import reactor.test.StepVerifier;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.BadSqlGrammarException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.messaging.Message;
import org.springframework.test.annotation.DirtiesContext;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
properties = "jdbc.supplier.query=select id, name from test order by id")
@SpringBootTest(properties = "jdbc.supplier.query=select id, name from test order by id")
@DirtiesContext
public class DefaultJdbcSupplierTests {
@@ -46,22 +43,22 @@ public class DefaultJdbcSupplierTests {
JdbcTemplate jdbcTemplate;
@Test
@SuppressWarnings({ "unchecked", "rawtypes" })
@SuppressWarnings("rawtypes")
void testExtraction() {
final Flux<Message<?>> messageFlux = jdbcSupplier.get();
StepVerifier stepVerifier = StepVerifier.create(messageFlux)
.assertNext((message) -> assertThat(message)
.satisfies((msg) -> assertThat(msg).extracting(Message::getPayload).matches(o -> {
.satisfies((msg) -> assertThat(msg).extracting(Message::getPayload).matches((o) -> {
Map map = (Map) o;
return map.get("ID").equals(1L) && map.get("NAME").equals("Bob");
})))
.assertNext((message) -> assertThat(message)
.satisfies((msg) -> assertThat(msg).extracting(Message::getPayload).matches(o -> {
.satisfies((msg) -> assertThat(msg).extracting(Message::getPayload).matches((o) -> {
Map map = (Map) o;
return map.get("ID").equals(2L) && map.get("NAME").equals("Jane");
})))
.assertNext((message) -> assertThat(message)
.satisfies((msg) -> assertThat(msg).extracting(Message::getPayload).matches(o -> {
.satisfies((msg) -> assertThat(msg).extracting(Message::getPayload).matches((o) -> {
Map map = (Map) o;
return map.get("ID").equals(3L) && map.get("NAME").equals("John");
})))
@@ -70,20 +67,6 @@ public class DefaultJdbcSupplierTests {
stepVerifier.verify();
}
/*
* The test to verify that DB is not initialized with Spring Integration DDL
* (spring.integration.jdbc.initialize-schema=NEVER) what happens by default via
* IntegrationAutoConfiguration.IntegrationJdbcConfiguration. This is not a
* functionality of this JDBC Supplier.
*/
@Test
void verifyNoIntMessageGroupTable() {
assertThatExceptionOfType(BadSqlGrammarException.class)
.isThrownBy(() -> this.jdbcTemplate.queryForList("SELECT * FROM INT_MESSAGE_GROUP"))
.havingCause()
.withMessageContaining("Table \"INT_MESSAGE_GROUP\" not found;");
}
@SpringBootApplication
static class JdbcSupplierTestApplication {

4
supplier/spring-jdbc-supplier/src/test/java/org/springframework/cloud/fn/supplier/jdbc/NonSplitJdbcSupplierTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2024 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.
@@ -34,7 +34,7 @@ import static org.assertj.core.api.Assertions.assertThat;
* @author Soby Chacko
* @author Artem Bilan
*/
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
@SpringBootTest(
properties = { "jdbc.supplier.query=select id, name from test order by id", "jdbc.supplier.split=false" })
@DirtiesContext
public class NonSplitJdbcSupplierTests {

2
supplier/spring-mongodb-supplier/README.adoc
View File

@@ -7,7 +7,7 @@ When you have use-cases such as periodical execution of querying MongoDB, based
## Beans for injection
You can import the `MongoDBSupplierConfiguration` in the application and then inject the following bean.
The `MongoDBSupplierConfiguration` auto-configuration provides the following bean:
`mongoDBSupplier`

23
supplier/spring-mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2024 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.
@@ -22,14 +22,14 @@ import java.util.function.Supplier;
import reactor.core.publisher.Flux;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.common.config.ComponentCustomizer;
import org.springframework.cloud.fn.splitter.SplitterFunctionConfiguration;
import org.springframework.cloud.function.context.PollableBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
@@ -38,17 +38,15 @@ import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
/**
* A configuration for MongoDB Source applications. Produces {@link MongoDbMessageSource}
* which polls collection with the query after startup according to the polling
* properties.
* Auto-configuration for MongoDB supplier. Produces {@link MongoDbMessageSource} which
* polls collection with the query after startup according to the polling properties.
*
* @author Adam Zwickey
* @author Artem Bilan
* @author David Turanski
*/
@Configuration(proxyBeanMethods = false)
@AutoConfiguration(after = { MongoAutoConfiguration.class, SplitterFunctionConfiguration.class })
@EnableConfigurationProperties({ MongodbSupplierProperties.class })
@Import(SplitterFunctionConfiguration.class)
public class MongodbSupplierConfiguration {
private final MongodbSupplierProperties properties;
@@ -69,9 +67,8 @@ public class MongodbSupplierConfiguration {
return () -> {
Message<?> received = mongoDbSource.receive();
if (received != null) {
return Flux.fromIterable(splitterFunction.apply(received)); // multiple
// Message<Map<String,
// Object>>
// multiple Message<Map<String, Object>>
return Flux.fromIterable(splitterFunction.apply(received));
}
else {
return Flux.empty();
@@ -89,8 +86,8 @@ public class MongodbSupplierConfiguration {
public MongoDbMessageSource mongoDbSource(
@Nullable ComponentCustomizer<MongoDbMessageSource> mongoDbMessageSourceCustomizer) {
Expression queryExpression = (this.properties.getQueryExpression() != null
? this.properties.getQueryExpression() : new LiteralExpression(this.properties.getQuery()));
Expression queryExpression = (this.properties.getQueryExpression() != null)
? this.properties.getQueryExpression() : new LiteralExpression(this.properties.getQuery());
MongoDbMessageSource mongoDbMessageSource = new MongoDbMessageSource(this.mongoTemplate, queryExpression);
mongoDbMessageSource.setCollectionNameExpression(new LiteralExpression(this.properties.getCollection()));
mongoDbMessageSource.setEntityClass(String.class);

12
supplier/spring-mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2024 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.
@@ -24,6 +24,8 @@ import org.springframework.expression.Expression;
import org.springframework.validation.annotation.Validated;
/**
* The MongoDB supplier configuration properties.
*
* @author Adam Zwickey
* @author Artem Bilan
* @author Chris Schaefer
@@ -60,7 +62,7 @@ public class MongodbSupplierProperties {
@NotEmpty(message = "Query is required")
public String getQuery() {
return query;
return this.query;
}
public void setQuery(String query) {
@@ -68,7 +70,7 @@ public class MongodbSupplierProperties {
}
public Expression getQueryExpression() {
return queryExpression;
return this.queryExpression;
}
public void setQueryExpression(Expression queryExpression) {
@@ -77,7 +79,7 @@ public class MongodbSupplierProperties {
@NotBlank(message = "Collection name is required")
public String getCollection() {
return collection;
return this.collection;
}
public void setCollection(String collection) {
@@ -85,7 +87,7 @@ public class MongodbSupplierProperties {
}
public boolean isSplit() {
return split;
return this.split;
}
public void setSplit(boolean split) {

4
supplier/spring-mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The MongoDB supplier auto-configuration support.
*/
package org.springframework.cloud.fn.supplier.mongo;

1
supplier/spring-mongodb-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.supplier.mongo.MongodbSupplierConfiguration

1
supplier/spring-mongodb-supplier/src/main/resources/application.properties
View File

@@ -1 +0,0 @@

19
supplier/spring-mongodb-supplier/src/test/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierApplicationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2023 the original author or authors.
* Copyright 2019-2024 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.
@@ -34,8 +34,8 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.fn.consumer.mongo.MongoDbTestContainerSupport;
import org.springframework.messaging.Message;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.util.ReflectionUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.entry;
@@ -43,14 +43,9 @@ import static org.assertj.core.api.Assertions.entry;
@SpringBootTest(
properties = { "mongodb.supplier.collection=testing", "mongodb.supplier.query={ name: { $exists: true }}",
"mongodb.supplier.update-expression='{ $unset: { name: 0 } }'" })
@DirtiesContext
class MongodbSupplierApplicationTests implements MongoDbTestContainerSupport {
@DynamicPropertySource
static void mongoDbProperties(DynamicPropertyRegistry registry) {
registry.add("spring.data.mongodb.port", MONGO_CONTAINER::getFirstMappedPort);
registry.add("spring.data.mongodb.database", () -> "test");
}
private final ObjectMapper objectMapper = new ObjectMapper();
@Autowired
@@ -60,7 +55,7 @@ class MongodbSupplierApplicationTests implements MongoDbTestContainerSupport {
private MongoClient mongo;
@BeforeEach
public void setUp() {
void setUp() {
MongoDatabase database = this.mongo.getDatabase("test");
database.createCollection("testing");
MongoCollection<Document> collection = database.getCollection("testing");
@@ -91,8 +86,8 @@ class MongodbSupplierApplicationTests implements MongoDbTestContainerSupport {
try {
map = objectMapper.readValue(message.getPayload().toString(), Map.class);
}
catch (Exception e) {
e.printStackTrace();
catch (Exception ex) {
ReflectionUtils.rethrowRuntimeException(ex);
}
return map;
}