Migrating pgcopy-sink
* pgcopy-sink is migrated from Spring Cloud Stream App Starters repository. * This is not converted as a function, rather preserved as a sink app. * Old location: https://github.com/spring-cloud-stream-app-starters/jdbc/tree/master/spring-cloud-starter-stream-sink-pgcopy
This commit is contained in:
Soby Chacko
58
applications/sink/pgcopy-sink/README.adoc
Normal file
58
applications/sink/pgcopy-sink/README.adoc
Normal file
@@ -0,0 +1,58 @@
|
||||
//tag::ref-doc[]
|
||||
= Pgcopy Sink
|
||||
|
||||
A module that writes its incoming payload to an RDBMS using the PostgreSQL COPY command.
|
||||
|
||||
== Input
|
||||
|
||||
=== Headers
|
||||
|
||||
=== Payload
|
||||
|
||||
* Any
|
||||
|
||||
Column expression will be evaluated against the message and the expression will usually be compatible with only one type (such as a Map or bean etc.)
|
||||
|
||||
== Output
|
||||
|
||||
N/A
|
||||
|
||||
== Options
|
||||
|
||||
The **$$jdbc$$** $$sink$$ has the following options:
|
||||
|
||||
//tag::configuration-properties[]
|
||||
$$spring.datasource.driver-class-name$$:: $$Fully qualified name of the JDBC driver. Auto-detected based on the URL by default.$$ *($$String$$, default: `$$<none>$$`)*
|
||||
$$spring.datasource.password$$:: $$Login password of the database.$$ *($$String$$, default: `$$<none>$$`)*
|
||||
$$spring.datasource.url$$:: $$JDBC URL of the database.$$ *($$String$$, default: `$$<none>$$`)*
|
||||
$$spring.datasource.username$$:: $$Login username of the database.$$ *($$String$$, default: `$$<none>$$`)*
|
||||
//end::configuration-properties[]
|
||||
|
||||
NOTE: The module also uses Spring Boot's https://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-sql.html#boot-features-configure-datasource[DataSource support] for configuring the database connection, so properties like `spring.datasource.url` _etc._ apply.
|
||||
|
||||
== Build
|
||||
|
||||
```
|
||||
$ ./mvnw clean install -PgenerateApps
|
||||
$ cd apps
|
||||
```
|
||||
You can find the corresponding binder based projects here.
|
||||
You can then cd into one one of the folders and build it:
|
||||
```
|
||||
$ ./mvnw clean package
|
||||
```
|
||||
|
||||
For integration tests to run, start a PostgreSQL database on localhost:
|
||||
|
||||
```
|
||||
docker run -e POSTGRES_PASSWORD=spring -e POSTGRES_DB=test -p 5432:5432 -d postgres:latest
|
||||
```
|
||||
|
||||
== Examples
|
||||
|
||||
```
|
||||
java -jar pgcopy-sink.jar --tableName=names --columns=name --spring.datasource.driver-class-name=org.mariadb.jdbc.Driver \
|
||||
--spring.datasource.url='jdbc:mysql://localhost:3306/test
|
||||
```
|
||||
|
||||
//end::ref-doc[]
|
||||
139
applications/sink/pgcopy-sink/pom.xml
Normal file
139
applications/sink/pgcopy-sink/pom.xml
Normal file
@@ -0,0 +1,139 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<artifactId>pgcopy-sink</artifactId>
|
||||
<version>3.0.0-SNAPSHOT</version>
|
||||
<name>pgcopy-sink</name>
|
||||
<description>pgcopy sink apps</description>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud.stream.app</groupId>
|
||||
<artifactId>stream-applications-core</artifactId>
|
||||
<version>3.0.0-SNAPSHOT</version>
|
||||
<relativePath/>
|
||||
</parent>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-jdbc</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-jdbc</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.postgresql</groupId>
|
||||
<artifactId>postgresql</artifactId>
|
||||
<scope>compile</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
|
||||
<version>${spring-cloud-stream.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-dataflow-apps-docs-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-dataflow-apps-generator-plugin</artifactId>
|
||||
<configuration>
|
||||
<application>
|
||||
<name>rabbit</name>
|
||||
<type>sink</type>
|
||||
<version>${project.version}</version>
|
||||
<configClass>org.springframework.cloud.fn.consumer.rabbit.RabbitConsumerConfiguration.class</configClass>
|
||||
|
||||
<maven>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.stream.app</groupId>
|
||||
<artifactId>pgcopy-sink</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</maven>
|
||||
</application>
|
||||
|
||||
</configuration>
|
||||
</plugin>
|
||||
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
<repositories>
|
||||
<repository>
|
||||
<snapshots>
|
||||
<enabled>true</enabled>
|
||||
</snapshots>
|
||||
<id>spring-snapshots</id>
|
||||
<name>Spring Snapshots</name>
|
||||
<url>https://repo.spring.io/libs-snapshot-local</url>
|
||||
</repository>
|
||||
<repository>
|
||||
<snapshots>
|
||||
<enabled>false</enabled>
|
||||
</snapshots>
|
||||
<id>spring-milestones</id>
|
||||
<name>Spring Milestones</name>
|
||||
<url>https://repo.spring.io/libs-milestone-local</url>
|
||||
</repository>
|
||||
<repository>
|
||||
<snapshots>
|
||||
<enabled>false</enabled>
|
||||
</snapshots>
|
||||
<id>spring-releases</id>
|
||||
<name>Spring Releases</name>
|
||||
<url>https://repo.spring.io/release</url>
|
||||
</repository>
|
||||
<repository>
|
||||
<snapshots>
|
||||
<enabled>false</enabled>
|
||||
</snapshots>
|
||||
<id>spring-libs-release</id>
|
||||
<name>Spring Libs Release</name>
|
||||
<url>https://repo.spring.io/libs-release</url>
|
||||
</repository>
|
||||
<repository>
|
||||
<snapshots>
|
||||
<enabled>false</enabled>
|
||||
</snapshots>
|
||||
<id>spring-milestone-release</id>
|
||||
<name>Spring Milestone Release</name>
|
||||
<url>https://repo.spring.io/libs-milestone</url>
|
||||
</repository>
|
||||
</repositories>
|
||||
<pluginRepositories>
|
||||
<pluginRepository>
|
||||
<id>spring-releases</id>
|
||||
<name>Spring Releases</name>
|
||||
<url>https://repo.spring.io/libs-release</url>
|
||||
</pluginRepository>
|
||||
<pluginRepository>
|
||||
<snapshots>
|
||||
<enabled>true</enabled>
|
||||
</snapshots>
|
||||
<id>spring-snapshots</id>
|
||||
<name>Spring Snapshots</name>
|
||||
<url>https://repo.spring.io/libs-snapshot-local</url>
|
||||
</pluginRepository>
|
||||
<pluginRepository>
|
||||
<snapshots>
|
||||
<enabled>false</enabled>
|
||||
</snapshots>
|
||||
<id>spring-milestones</id>
|
||||
<name>Spring Milestones</name>
|
||||
<url>https://repo.spring.io/libs-milestone-local</url>
|
||||
</pluginRepository>
|
||||
</pluginRepositories>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,59 @@
|
||||
/*
|
||||
* Copyright 2016-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.stream.app.pgcopy.sink;
|
||||
|
||||
import java.nio.charset.Charset;
|
||||
import java.util.Collection;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.core.io.ByteArrayResource;
|
||||
|
||||
/**
|
||||
* An in-memory script crafted for dropping-creating the table we're working with.
|
||||
* All columns are created as VARCHAR(2000).
|
||||
*
|
||||
* @author Eric Bottard
|
||||
* @author Thomas Risberg
|
||||
*/
|
||||
public class DefaultInitializationScriptResource extends ByteArrayResource {
|
||||
|
||||
private static final Log logger = LogFactory.getLog(DefaultInitializationScriptResource.class);
|
||||
|
||||
public DefaultInitializationScriptResource(String tableName, Collection<String> columns) {
|
||||
super(scriptFor(tableName, columns).getBytes(Charset.forName("UTF-8")));
|
||||
}
|
||||
|
||||
private static String scriptFor(String tableName, Collection<String> columns) {
|
||||
StringBuilder result = new StringBuilder("DROP TABLE ");
|
||||
result.append(tableName).append(";\n\n");
|
||||
|
||||
result.append("CREATE TABLE ").append(tableName).append('(');
|
||||
int i = 0;
|
||||
for (String column : columns) {
|
||||
if (i++ > 0) {
|
||||
result.append(", ");
|
||||
}
|
||||
result.append(column).append(" VARCHAR(2000)");
|
||||
}
|
||||
result.append(");\n");
|
||||
logger.debug(String.format("Generated the following initializing script for table %s:\n%s", tableName,
|
||||
result.toString()));
|
||||
return result.toString();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,332 @@
|
||||
/*
|
||||
* Copyright 2016-2019 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.stream.app.pgcopy.sink;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
|
||||
import javax.annotation.PreDestroy;
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.postgresql.copy.CopyIn;
|
||||
import org.postgresql.copy.CopyManager;
|
||||
import org.postgresql.core.BaseConnection;
|
||||
|
||||
import org.springframework.beans.factory.FactoryBean;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
|
||||
import org.springframework.cloud.stream.messaging.Sink;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.core.io.ResourceLoader;
|
||||
import org.springframework.dao.DataAccessException;
|
||||
import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor;
|
||||
import org.springframework.integration.aggregator.ExpressionEvaluatingCorrelationStrategy;
|
||||
import org.springframework.integration.aggregator.MessageCountReleaseStrategy;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.config.AggregatorFactoryBean;
|
||||
import org.springframework.integration.store.MessageGroupStore;
|
||||
import org.springframework.integration.store.MessageGroupStoreReaper;
|
||||
import org.springframework.integration.store.SimpleMessageStore;
|
||||
import org.springframework.jdbc.core.ConnectionCallback;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.datasource.init.DataSourceInitializer;
|
||||
import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.scheduling.annotation.EnableScheduling;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.TransactionStatus;
|
||||
import org.springframework.transaction.support.TransactionCallback;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* Configuration class for the PostgreSQL CopyManager.
|
||||
*
|
||||
* @author Thomas Risberg
|
||||
* @author Janne Valkealahti
|
||||
*/
|
||||
@Configuration
|
||||
@EnableScheduling
|
||||
@EnableBinding(Sink.class)
|
||||
@EnableConfigurationProperties(PgcopySinkProperties.class)
|
||||
public class PgcopySinkConfiguration {
|
||||
|
||||
private static final Log logger = LogFactory.getLog(PgcopySinkConfiguration.class);
|
||||
|
||||
@Autowired
|
||||
private PgcopySinkProperties properties;
|
||||
|
||||
@Bean
|
||||
public MessageChannel toSink() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
@Primary
|
||||
@ServiceActivator(inputChannel = Sink.INPUT)
|
||||
FactoryBean<MessageHandler> aggregatorFactoryBean(MessageChannel toSink, MessageGroupStore messageGroupStore) {
|
||||
AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
|
||||
aggregatorFactoryBean.setCorrelationStrategy(
|
||||
new ExpressionEvaluatingCorrelationStrategy("payload.getClass().name"));
|
||||
aggregatorFactoryBean.setReleaseStrategy(new MessageCountReleaseStrategy(properties.getBatchSize()));
|
||||
aggregatorFactoryBean.setMessageStore(messageGroupStore);
|
||||
aggregatorFactoryBean.setProcessorBean(new DefaultAggregatingMessageGroupProcessor());
|
||||
aggregatorFactoryBean.setExpireGroupsUponCompletion(true);
|
||||
aggregatorFactoryBean.setSendPartialResultOnExpiry(true);
|
||||
aggregatorFactoryBean.setOutputChannel(toSink);
|
||||
return aggregatorFactoryBean;
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ServiceActivator(inputChannel = "toSink")
|
||||
public MessageHandler datasetSinkMessageHandler(final JdbcTemplate jdbcTemplate,
|
||||
final PlatformTransactionManager platformTransactionManager) {
|
||||
|
||||
final TransactionTemplate txTemplate = new TransactionTemplate(platformTransactionManager);
|
||||
|
||||
if (StringUtils.hasText(properties.getErrorTable())) {
|
||||
verifyErrorTable(jdbcTemplate, txTemplate);
|
||||
}
|
||||
|
||||
StringBuilder columns = new StringBuilder();
|
||||
for (String col : properties.getColumns()) {
|
||||
if (columns.length() > 0) {
|
||||
columns.append(",");
|
||||
}
|
||||
columns.append(col);
|
||||
}
|
||||
// the copy command
|
||||
final StringBuilder sql = new StringBuilder("COPY " + properties.getTableName());
|
||||
if (columns.length() > 0) {
|
||||
sql.append(" (" + columns + ")");
|
||||
}
|
||||
sql.append(" FROM STDIN");
|
||||
|
||||
StringBuilder options = new StringBuilder();
|
||||
if (properties.getFormat() == PgcopySinkProperties.Format.CSV) {
|
||||
options.append("CSV");
|
||||
}
|
||||
if (properties.getDelimiter() != null) {
|
||||
options.append(escapedOptionCharacterValue(options.length(), "DELIMITER", properties.getDelimiter()));
|
||||
}
|
||||
if (properties.getNullString() != null) {
|
||||
options.append((options.length() > 0 ? " " : "") + "NULL '" + properties.getNullString() + "'");
|
||||
}
|
||||
if (properties.getQuote() != null) {
|
||||
options.append(quotedOptionCharacterValue(options.length(), "QUOTE", properties.getQuote()));
|
||||
}
|
||||
if (properties.getEscape() != null) {
|
||||
options.append(quotedOptionCharacterValue(options.length(), "ESCAPE", properties.getEscape()));
|
||||
}
|
||||
if (options.length() > 0) {
|
||||
sql.append(" WITH " + options.toString());
|
||||
}
|
||||
|
||||
return new MessageHandler() {
|
||||
|
||||
@Override
|
||||
public void handleMessage(Message<?> message) throws MessagingException {
|
||||
Object payload = message.getPayload();
|
||||
if (payload instanceof Collection<?>) {
|
||||
final Collection<?> payloads = (Collection<?>) payload;
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Executing batch of size " + payloads.size() + " for " + sql);
|
||||
}
|
||||
try {
|
||||
long rows = doCopy(payloads, txTemplate);
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Wrote " + rows + " rows");
|
||||
}
|
||||
}
|
||||
catch (DataAccessException e) {
|
||||
logger.error("Error while copying batch of data: " + e.getMessage());
|
||||
logger.error("Switching to single row copy for current batch");
|
||||
long rows = 0;
|
||||
for (Object singlePayload : payloads) {
|
||||
try {
|
||||
rows = rows + doCopy(Collections.singletonList(singlePayload), txTemplate);
|
||||
}
|
||||
catch (DataAccessException e2) {
|
||||
logger.error("Copy for single row caused error: " + e2.getMessage());
|
||||
logger.error("Bad Data: \n" + singlePayload);
|
||||
if (StringUtils.hasText(properties.getErrorTable())) {
|
||||
writeError(e2, singlePayload);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Re-tried batch and wrote " + rows + " rows");
|
||||
}
|
||||
}
|
||||
}
|
||||
else {
|
||||
throw new IllegalStateException("Expected a collection of strings but received " +
|
||||
message.getPayload().getClass().getName());
|
||||
}
|
||||
}
|
||||
|
||||
private void writeError(final DataAccessException exception, final Object payload) {
|
||||
final String message;
|
||||
if (exception.getCause() != null) {
|
||||
message = exception.getCause().getMessage();
|
||||
}
|
||||
else {
|
||||
message = exception.getMessage();
|
||||
}
|
||||
try {
|
||||
txTemplate.execute(new TransactionCallback<Long>() {
|
||||
@Override
|
||||
public Long doInTransaction(TransactionStatus transactionStatus) {
|
||||
jdbcTemplate.update(
|
||||
"insert into " + properties.getErrorTable() + " (table_name, error_message, payload) values (?, ?, ?)",
|
||||
new Object[]{properties.getTableName(), message, payload});
|
||||
return null;
|
||||
}
|
||||
});
|
||||
}
|
||||
catch (DataAccessException e) {
|
||||
logger.error("Writing to error table failed: " + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private long doCopy(final Collection<?> payloads, TransactionTemplate txTemplate) {
|
||||
Long rows = txTemplate.execute(transactionStatus -> jdbcTemplate.execute(
|
||||
new ConnectionCallback<Long>() {
|
||||
@Override
|
||||
public Long doInConnection(Connection connection) throws SQLException, DataAccessException {
|
||||
CopyManager cm = connection.unwrap(BaseConnection.class).getCopyAPI();
|
||||
CopyIn ci = cm.copyIn(sql.toString());
|
||||
for (Object payloadData : payloads) {
|
||||
String textPayload = (payloadData instanceof byte[]) ?
|
||||
new String((byte[]) payloadData) : (String) payloadData;
|
||||
byte[] data = (textPayload + "\n").getBytes();
|
||||
ci.writeToCopy(data, 0, data.length);
|
||||
}
|
||||
return Long.valueOf(ci.endCopy());
|
||||
}
|
||||
}
|
||||
));
|
||||
return rows;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@ConditionalOnProperty("pgcopy.initialize")
|
||||
@Bean
|
||||
public DataSourceInitializer nonBootDataSourceInitializer(DataSource dataSource, ResourceLoader resourceLoader) {
|
||||
DataSourceInitializer dataSourceInitializer = new DataSourceInitializer();
|
||||
dataSourceInitializer.setDataSource(dataSource);
|
||||
ResourceDatabasePopulator databasePopulator = new ResourceDatabasePopulator();
|
||||
databasePopulator.setIgnoreFailedDrops(true);
|
||||
dataSourceInitializer.setDatabasePopulator(databasePopulator);
|
||||
if ("true".equals(properties.getInitialize())) {
|
||||
databasePopulator.addScript(new DefaultInitializationScriptResource(properties.getTableName(),
|
||||
properties.getColumns()));
|
||||
}
|
||||
else {
|
||||
databasePopulator.addScript(resourceLoader.getResource(properties.getInitialize()));
|
||||
}
|
||||
return dataSourceInitializer;
|
||||
}
|
||||
|
||||
@Bean
|
||||
MessageGroupStore messageGroupStore() {
|
||||
SimpleMessageStore messageGroupStore = new SimpleMessageStore();
|
||||
messageGroupStore.setTimeoutOnIdle(true);
|
||||
messageGroupStore.setCopyOnGet(false);
|
||||
return messageGroupStore;
|
||||
}
|
||||
|
||||
@Bean
|
||||
MessageGroupStoreReaper messageGroupStoreReaper(MessageGroupStore messageStore,
|
||||
InputBindingLifecycle inputBindingLifecycle) {
|
||||
MessageGroupStoreReaper messageGroupStoreReaper = new MessageGroupStoreReaper(messageStore);
|
||||
messageGroupStoreReaper.setPhase(inputBindingLifecycle.getPhase() - 1);
|
||||
messageGroupStoreReaper.setTimeout(properties.getIdleTimeout());
|
||||
messageGroupStoreReaper.setAutoStartup(true);
|
||||
messageGroupStoreReaper.setExpireOnDestroy(true);
|
||||
return messageGroupStoreReaper;
|
||||
}
|
||||
|
||||
@Bean
|
||||
ReaperTask reaperTask() {
|
||||
return new ReaperTask();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public JdbcTemplate jdbcTemplate(DataSource dataSource) {
|
||||
JdbcTemplate jt = new JdbcTemplate(dataSource);
|
||||
return jt;
|
||||
}
|
||||
|
||||
private String quotedOptionCharacterValue(int length, String option, char value) {
|
||||
return (length > 0 ? " " : "") + option + " '" + (value == '\'' ? "''" : value) + "'";
|
||||
}
|
||||
|
||||
private String escapedOptionCharacterValue(int length, String option, String value) {
|
||||
return (length > 0 ? " " : "") + option + " " + (value.startsWith("\\") ? "E'" + value : "'" + value) + "'";
|
||||
}
|
||||
|
||||
private void verifyErrorTable(final JdbcTemplate jdbcTemplate, final TransactionTemplate txTemplate) {
|
||||
try {
|
||||
txTemplate.execute(new TransactionCallback<Long>() {
|
||||
@Override
|
||||
public Long doInTransaction(TransactionStatus transactionStatus) {
|
||||
jdbcTemplate.update(
|
||||
"insert into " + properties.getErrorTable() + " (table_name, error_message, payload) values (?, ?, ?)",
|
||||
properties.getErrorTable(), "message", "payload");
|
||||
transactionStatus.setRollbackOnly();
|
||||
return null;
|
||||
}
|
||||
});
|
||||
}
|
||||
catch (DataAccessException e) {
|
||||
throw new IllegalStateException("Invalid error table specified", e);
|
||||
}
|
||||
}
|
||||
|
||||
public static class ReaperTask {
|
||||
|
||||
@Autowired
|
||||
MessageGroupStoreReaper messageGroupStoreReaper;
|
||||
|
||||
@Scheduled(fixedRate = 1000)
|
||||
public void reap() {
|
||||
messageGroupStoreReaper.run();
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
public void beforeDestroy() {
|
||||
reap();
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,204 @@
|
||||
/*
|
||||
* Copyright 2016-2017 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.stream.app.pgcopy.sink;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
import javax.validation.constraints.NotNull;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
|
||||
/**
|
||||
* Used to configure the pgcopy sink module options that are related to writing using the PostgreSQL CopyManager API.
|
||||
*
|
||||
* @author Thomas Risberg
|
||||
*/
|
||||
@SuppressWarnings("unused")
|
||||
@ConfigurationProperties("pgcopy")
|
||||
@Validated
|
||||
public class PgcopySinkProperties {
|
||||
|
||||
/**
|
||||
* The name of the table to write into.
|
||||
*/
|
||||
@NotNull
|
||||
private String tableName;
|
||||
|
||||
/**
|
||||
* The names of the columns that shall receive data.
|
||||
* Also used at initialization time to issue the DDL.
|
||||
*/
|
||||
private List<String> columns = Collections.singletonList("payload");
|
||||
|
||||
/**
|
||||
* Threshold in number of messages when data will be flushed to database table.
|
||||
*/
|
||||
private int batchSize = 10000;
|
||||
|
||||
/**
|
||||
* Idle timeout in milliseconds when data is automatically flushed to database table.
|
||||
*/
|
||||
private long idleTimeout = -1L;
|
||||
|
||||
/**
|
||||
* 'true', 'false' or the location of a custom initialization script for the table.
|
||||
*/
|
||||
private String initialize = "false";
|
||||
|
||||
/**
|
||||
* Format to use for the copy command.
|
||||
*/
|
||||
private Format format = Format.TEXT;
|
||||
|
||||
/**
|
||||
* Specifies the string that represents a null value. The default is \N (backslash-N) in text format, and an
|
||||
* unquoted empty string in CSV format.
|
||||
*/
|
||||
private String nullString;
|
||||
|
||||
/**
|
||||
* Specifies the character that separates columns within each row (line) of the file. The default is a tab character
|
||||
* in text format, a comma in CSV format. This must be a single one-byte character. Using an escaped value like '\t'
|
||||
* is allowed.
|
||||
*/
|
||||
private String delimiter;
|
||||
|
||||
/**
|
||||
* Specifies the quoting character to be used when a data value is quoted. The default is double-quote. This must
|
||||
* be a single one-byte character. This option is allowed only when using CSV format.
|
||||
*/
|
||||
private Character quote;
|
||||
|
||||
/**
|
||||
* Specifies the character that should appear before a data character that matches the QUOTE value. The default is
|
||||
* the same as the QUOTE value (so that the quoting character is doubled if it appears in the data). This must be
|
||||
* a single one-byte character. This option is allowed only when using CSV format.
|
||||
*/
|
||||
private Character escape;
|
||||
|
||||
/**
|
||||
* The name of the error table used for writing rows causing errors. The error table should have three columns
|
||||
* named "table_name", "error_message" and "payload" large enough to hold potential data values.
|
||||
* You can use the following DDL to create this table:
|
||||
* 'CREATE TABLE ERRORS (TABLE_NAME VARCHAR(255), ERROR_MESSAGE TEXT,PAYLOAD TEXT)'
|
||||
*/
|
||||
private String errorTable;
|
||||
|
||||
|
||||
public String getTableName() {
|
||||
return tableName;
|
||||
}
|
||||
|
||||
public void setTableName(String tableName) {
|
||||
this.tableName = tableName;
|
||||
}
|
||||
|
||||
public List<String> getColumns() {
|
||||
return columns;
|
||||
}
|
||||
|
||||
public void setColumns(List<String> columns) {
|
||||
this.columns = columns;
|
||||
}
|
||||
|
||||
public int getBatchSize() {
|
||||
return batchSize;
|
||||
}
|
||||
|
||||
public void setBatchSize(int batchSize) {
|
||||
this.batchSize = batchSize;
|
||||
}
|
||||
|
||||
public long getIdleTimeout() {
|
||||
return idleTimeout;
|
||||
}
|
||||
|
||||
public void setIdleTimeout(long idleTimeout) {
|
||||
this.idleTimeout = idleTimeout;
|
||||
}
|
||||
|
||||
public String getInitialize() {
|
||||
return initialize;
|
||||
}
|
||||
|
||||
public void setInitialize(String initialize) {
|
||||
this.initialize = initialize;
|
||||
}
|
||||
|
||||
public Format getFormat() {
|
||||
return format;
|
||||
}
|
||||
|
||||
public void setFormat(Format format) {
|
||||
this.format = format;
|
||||
}
|
||||
|
||||
public String getNullString() {
|
||||
return nullString;
|
||||
}
|
||||
|
||||
public void setNullString(String nullString) {
|
||||
this.nullString = nullString;
|
||||
}
|
||||
|
||||
public String getDelimiter() {
|
||||
return delimiter;
|
||||
}
|
||||
|
||||
public void setDelimiter(String delimiter) {
|
||||
this.delimiter = delimiter;
|
||||
}
|
||||
|
||||
public Character getQuote() {
|
||||
return quote;
|
||||
}
|
||||
|
||||
public void setQuote(Character quote) {
|
||||
this.quote = quote;
|
||||
}
|
||||
|
||||
public Character getEscape() {
|
||||
return escape;
|
||||
}
|
||||
|
||||
public void setEscape(Character escape) {
|
||||
this.escape = escape;
|
||||
}
|
||||
|
||||
public String getErrorTable() {
|
||||
return errorTable;
|
||||
}
|
||||
|
||||
public void setErrorTable(String errorTable) {
|
||||
this.errorTable = errorTable;
|
||||
}
|
||||
|
||||
public enum Format {
|
||||
|
||||
/**
|
||||
* text.
|
||||
*/
|
||||
TEXT,
|
||||
/**
|
||||
* csv.
|
||||
*/
|
||||
CSV
|
||||
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
configuration-properties.classes=org.springframework.cloud.stream.app.pgcopy.sink.PgcopySinkProperties
|
||||
configuration-properties.names=\
|
||||
spring.datasource.url,\
|
||||
spring.datasource.driver-class-name,\
|
||||
spring.datasource.username,\
|
||||
spring.datasource.password
|
||||
@@ -0,0 +1,6 @@
|
||||
configuration-properties.classes=org.springframework.cloud.stream.app.pgcopy.sink.PgcopySinkProperties
|
||||
configuration-properties.names=\
|
||||
spring.datasource.url,\
|
||||
spring.datasource.driver-class-name,\
|
||||
spring.datasource.username,\
|
||||
spring.datasource.password
|
||||
@@ -0,0 +1 @@
|
||||
provides: spring-cloud-starter-stream-sink-pgcopy
|
||||
@@ -0,0 +1,170 @@
|
||||
/*
|
||||
* Copyright 2016-2019 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.stream.app.pgcopy.sink;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Properties;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Before;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Test;
|
||||
import org.postgresql.util.PSQLException;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.test.util.TestPropertyValues;
|
||||
import org.springframework.cloud.stream.app.pgcopy.test.PostgresTestSupport;
|
||||
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
|
||||
import org.springframework.core.io.support.PropertiesLoaderUtils;
|
||||
import org.springframework.dao.DataAccessException;
|
||||
import org.springframework.jdbc.core.JdbcOperations;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.datasource.DriverManagerDataSource;
|
||||
|
||||
import static org.hamcrest.Matchers.is;
|
||||
|
||||
/**
|
||||
* Integration Tests testing bad error table specified for PgcopySink. Only runs if PostgreSQL database is available.
|
||||
*
|
||||
* @author Thomas Risberg
|
||||
* @author Artem Bilan
|
||||
*/
|
||||
public class PgcopyBadErrorTableIntegrationTests {
|
||||
|
||||
@ClassRule
|
||||
public static PostgresTestSupport postgresAvailable = new PostgresTestSupport();
|
||||
|
||||
private String[] env = { "pgcopy.tableName=names", "pgcopy.columns=id,name,age", "pgcopy.format=CSV" };
|
||||
|
||||
private String[] jdbc = { };
|
||||
|
||||
private Properties appProperties = new Properties();
|
||||
|
||||
@Before
|
||||
public void setup() {
|
||||
try {
|
||||
appProperties = PropertiesLoaderUtils.loadAllProperties("application.properties");
|
||||
}
|
||||
catch (IOException e) {
|
||||
}
|
||||
List<String> jdbcProperties = new ArrayList<>();
|
||||
for (Object key : appProperties.keySet()) {
|
||||
jdbcProperties.add(key + "=" + appProperties.get(key));
|
||||
}
|
||||
this.jdbc = jdbcProperties.toArray(new String[0]);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBadErrorTableName() {
|
||||
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
|
||||
TestPropertyValues.of(this.jdbc)
|
||||
.and("pgcopy.error-table=missing")
|
||||
.and(this.env)
|
||||
.applyTo(context);
|
||||
context.register(PgcopySinkApplication.class);
|
||||
try {
|
||||
context.refresh();
|
||||
}
|
||||
catch (Exception e) {
|
||||
Throwable ise = null;
|
||||
Throwable dae = null;
|
||||
Throwable cause = e;
|
||||
while (cause.getCause() != null) {
|
||||
cause = cause.getCause();
|
||||
if (cause instanceof IllegalStateException) {
|
||||
ise = cause;
|
||||
}
|
||||
if (cause instanceof DataAccessException) {
|
||||
dae = cause;
|
||||
}
|
||||
}
|
||||
Assert.assertThat(cause.getClass().getName(), is(PSQLException.class.getName()));
|
||||
Assert.assertNotNull(ise);
|
||||
Assert.assertTrue(ise.getMessage().contains("Invalid error table specified"));
|
||||
Assert.assertNotNull(dae);
|
||||
Assert.assertTrue(cause.getMessage().contains("relation"));
|
||||
Assert.assertTrue(cause.getMessage().contains("does not exist"));
|
||||
}
|
||||
context.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBadErrorTableFields() {
|
||||
DriverManagerDataSource dataSource = new DriverManagerDataSource();
|
||||
dataSource.setDriverClassName(appProperties.getProperty("spring.datasource.driver-class-name"));
|
||||
dataSource.setUrl(appProperties.getProperty("spring.datasource.url"));
|
||||
dataSource.setUsername(appProperties.getProperty("spring.datasource.username"));
|
||||
dataSource.setPassword(appProperties.getProperty("spring.datasource.password"));
|
||||
JdbcOperations jdbcOperations = new JdbcTemplate(dataSource);
|
||||
try {
|
||||
jdbcOperations.execute(
|
||||
"drop table test_errors");
|
||||
}
|
||||
catch (Exception e) {
|
||||
}
|
||||
try {
|
||||
jdbcOperations.execute(
|
||||
"create table test_errors (table_name varchar(255), error_message text)");
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new IllegalStateException("Error creating table", e);
|
||||
}
|
||||
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
|
||||
TestPropertyValues.of(this.jdbc)
|
||||
.and("pgcopy.error-table=test_errors")
|
||||
.and(this.env)
|
||||
.applyTo(context);
|
||||
context.register(PgcopySinkApplication.class);
|
||||
try {
|
||||
context.refresh();
|
||||
}
|
||||
catch (Exception e) {
|
||||
Throwable ise = null;
|
||||
Throwable dae = null;
|
||||
Throwable cause = e;
|
||||
while (cause.getCause() != null) {
|
||||
cause = cause.getCause();
|
||||
if (cause instanceof IllegalStateException) {
|
||||
ise = cause;
|
||||
}
|
||||
if (cause instanceof DataAccessException) {
|
||||
dae = cause;
|
||||
}
|
||||
}
|
||||
Assert.assertThat(cause.getClass().getName(), is(PSQLException.class.getName()));
|
||||
Assert.assertNotNull(ise);
|
||||
Assert.assertTrue(ise.getMessage().contains("Invalid error table specified"));
|
||||
Assert.assertNotNull(dae);
|
||||
Assert.assertTrue(cause.getMessage().contains("column"));
|
||||
Assert.assertTrue(cause.getMessage().contains("does not exist"));
|
||||
}
|
||||
context.close();
|
||||
}
|
||||
|
||||
@SpringBootApplication
|
||||
public static class PgcopySinkApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(PgcopySinkApplication.class, args);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
/*
|
||||
* Copyright 2016-2019 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.stream.app.pgcopy.sink;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.cloud.stream.app.pgcopy.test.PostgresTestSupport;
|
||||
import org.springframework.cloud.stream.messaging.Sink;
|
||||
import org.springframework.jdbc.core.JdbcOperations;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.TestPropertySource;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
|
||||
import static org.hamcrest.Matchers.is;
|
||||
|
||||
/**
|
||||
* Integration Tests for PgcopySink with error table. Only runs if PostgreSQL database is available.
|
||||
*
|
||||
* @author Thomas Risberg
|
||||
* @author Janne Valkealahti
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
|
||||
classes = PgcopyErrorTableIntegrationTests.PgcopySinkApplication.class)
|
||||
@TestPropertySource(properties = { "pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true",
|
||||
"pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.error-table=test_errors",
|
||||
"spring.datasource.initialization-mode=always", "spring.datasource.schema=classpath:error-table-ddl.sql",
|
||||
"spring.datasource.continue-on-error=true" })
|
||||
@DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_EACH_TEST_METHOD)
|
||||
public class PgcopyErrorTableIntegrationTests {
|
||||
|
||||
@ClassRule
|
||||
public static PostgresTestSupport postgresAvailable = new PostgresTestSupport();
|
||||
|
||||
@Autowired
|
||||
protected Sink channels;
|
||||
|
||||
@Autowired
|
||||
protected JdbcOperations jdbcOperations;
|
||||
|
||||
@Test
|
||||
public void testCopyCSV() {
|
||||
channels.input().send(MessageBuilder.withPayload("123,Nisse,25").build());
|
||||
channels.input().send(MessageBuilder.withPayload("GARBAGE").build());
|
||||
channels.input().send(MessageBuilder.withPayload("125,Bubba,22").build());
|
||||
int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class);
|
||||
int errors = jdbcOperations.queryForObject("select count(*) from test_errors", Integer.class);
|
||||
Assert.assertThat(result, is(2));
|
||||
Assert.assertThat(errors, is(1));
|
||||
}
|
||||
|
||||
@SpringBootApplication
|
||||
public static class PgcopySinkApplication {
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(PgcopySinkApplication.class, args);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
/*
|
||||
* Copyright 2016-2019 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.stream.app.pgcopy.sink;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.cloud.stream.app.pgcopy.test.PostgresTestSupport;
|
||||
import org.springframework.cloud.stream.messaging.Sink;
|
||||
import org.springframework.jdbc.core.JdbcOperations;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.TestPropertySource;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
|
||||
import static org.hamcrest.Matchers.is;
|
||||
|
||||
/**
|
||||
* Integration Tests for PgcopySink. Only runs if PostgreSQL database is available.
|
||||
*
|
||||
* @author Thomas Risberg
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
|
||||
classes = PgcopySinkIntegrationTests.PgcopySinkApplication.class)
|
||||
@DirtiesContext
|
||||
public abstract class PgcopySinkIntegrationTests {
|
||||
|
||||
@ClassRule
|
||||
public static PostgresTestSupport postgresAvailable = new PostgresTestSupport();
|
||||
|
||||
@Autowired
|
||||
protected Sink channels;
|
||||
|
||||
@Autowired
|
||||
protected JdbcOperations jdbcOperations;
|
||||
|
||||
@TestPropertySource(properties = {"pgcopy.table-name=test", "pgcopy.batch-size=1", "pgcopy.initialize=true"})
|
||||
public static class BasicPayloadCopyTests extends PgcopySinkIntegrationTests {
|
||||
|
||||
@Test
|
||||
public void testBasicCopy() {
|
||||
String sent = "hello42";
|
||||
channels.input().send(MessageBuilder.withPayload(sent).build());
|
||||
String result = jdbcOperations.queryForObject("select payload from test", String.class);
|
||||
Assert.assertThat(result, is("hello42"));
|
||||
}
|
||||
}
|
||||
|
||||
@TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=4", "pgcopy.initialize=true",
|
||||
"pgcopy.columns=id,name,age"})
|
||||
public static class PgcopyTextTests extends PgcopySinkIntegrationTests {
|
||||
|
||||
@Test
|
||||
public void testCopyText() {
|
||||
channels.input().send(MessageBuilder.withPayload("123\tNisse\t25").build());
|
||||
channels.input().send(MessageBuilder.withPayload("124\tAnna\t21").build());
|
||||
channels.input().send(MessageBuilder.withPayload("125\tBubba\t22").build());
|
||||
channels.input().send(MessageBuilder.withPayload("126\tPelle\t32").build());
|
||||
int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class);
|
||||
Assert.assertThat(result, is(4));
|
||||
}
|
||||
}
|
||||
|
||||
@TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true",
|
||||
"pgcopy.columns=id,name,age", "pgcopy.format=CSV"})
|
||||
public static class PgcopyCSVTests extends PgcopySinkIntegrationTests {
|
||||
|
||||
@Test
|
||||
public void testCopyCSV() {
|
||||
channels.input().send(MessageBuilder.withPayload("123,\"Nisse\",25").build());
|
||||
channels.input().send(MessageBuilder.withPayload("124,\"Anna\",21").build());
|
||||
channels.input().send(MessageBuilder.withPayload("125,\"Bubba\",22").build());
|
||||
int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class);
|
||||
Assert.assertThat(result, is(3));
|
||||
}
|
||||
}
|
||||
|
||||
@TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true",
|
||||
"pgcopy.columns=id,name,age", "pgcopy.format=CSV"})
|
||||
public static class PgcopyNullTests extends PgcopySinkIntegrationTests {
|
||||
|
||||
@Test
|
||||
public void testCopyCSV() {
|
||||
channels.input().send(MessageBuilder.withPayload("123,\"Nisse\",25").build());
|
||||
channels.input().send(MessageBuilder.withPayload("124,,21").build());
|
||||
channels.input().send(MessageBuilder.withPayload("125,\"Bubba\",22").build());
|
||||
int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class);
|
||||
int nulls = jdbcOperations.queryForObject("select count(*) from names where name is null", Integer.class);
|
||||
Assert.assertThat(result, is(3));
|
||||
Assert.assertThat(nulls, is(1));
|
||||
}
|
||||
}
|
||||
|
||||
@TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true",
|
||||
"pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.null-string=null"})
|
||||
public static class PgcopyNullStringTests extends PgcopySinkIntegrationTests {
|
||||
|
||||
@Test
|
||||
public void testCopyCSV() {
|
||||
channels.input().send(MessageBuilder.withPayload("123,\"Nisse\",25").build());
|
||||
channels.input().send(MessageBuilder.withPayload("124,null,21").build());
|
||||
channels.input().send(MessageBuilder.withPayload("125,\"Bubba\",22").build());
|
||||
int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class);
|
||||
int nulls = jdbcOperations.queryForObject("select count(*) from names where name is null", Integer.class);
|
||||
Assert.assertThat(result, is(3));
|
||||
Assert.assertThat(nulls, is(1));
|
||||
}
|
||||
}
|
||||
|
||||
@TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true",
|
||||
"pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.delimiter=|"})
|
||||
public static class PgcopyDelimiterTests extends PgcopySinkIntegrationTests {
|
||||
|
||||
@Test
|
||||
public void testCopyCSV() {
|
||||
channels.input().send(MessageBuilder.withPayload("123|\"Nisse\"|25").build());
|
||||
channels.input().send(MessageBuilder.withPayload("124|\"Anna\"|21").build());
|
||||
channels.input().send(MessageBuilder.withPayload("125|\"Bubba\"|22").build());
|
||||
int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class);
|
||||
Assert.assertThat(result, is(3));
|
||||
}
|
||||
}
|
||||
|
||||
@TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true",
|
||||
"pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.delimiter=\\t"})
|
||||
public static class PgcopyEscapedDelimiterTests extends PgcopySinkIntegrationTests {
|
||||
|
||||
@Test
|
||||
public void testCopyCSV() {
|
||||
channels.input().send(MessageBuilder.withPayload("123\t\"Nisse\"\t25").build());
|
||||
channels.input().send(MessageBuilder.withPayload("124\t\"Anna\"\t21").build());
|
||||
channels.input().send(MessageBuilder.withPayload("125\t\"Bubba\"\t22").build());
|
||||
int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class);
|
||||
Assert.assertThat(result, is(3));
|
||||
}
|
||||
}
|
||||
|
||||
@TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true",
|
||||
"pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.quote='"})
|
||||
public static class PgcopyQuoteTests extends PgcopySinkIntegrationTests {
|
||||
|
||||
@Test
|
||||
public void testCopyCSV() {
|
||||
channels.input().send(MessageBuilder.withPayload("123,Nisse,25").build());
|
||||
channels.input().send(MessageBuilder.withPayload("124,'Anna',21").build());
|
||||
channels.input().send(MessageBuilder.withPayload("125,Bubba,22").build());
|
||||
int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class);
|
||||
int quoted = jdbcOperations.queryForObject("select count(*) from names where name = 'Anna'", Integer.class);
|
||||
Assert.assertThat(result, is(3));
|
||||
Assert.assertThat(quoted, is(1));
|
||||
}
|
||||
}
|
||||
|
||||
@TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true",
|
||||
"pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.escape=\\\\"})
|
||||
public static class PgcopyEscapeTests extends PgcopySinkIntegrationTests {
|
||||
|
||||
@Test
|
||||
public void testCopyCSV() {
|
||||
channels.input().send(MessageBuilder.withPayload("123,Nisse,25").build());
|
||||
channels.input().send(MessageBuilder.withPayload("124,\"Anna\\\"\",21").build());
|
||||
channels.input().send(MessageBuilder.withPayload("125,Bubba,22").build());
|
||||
int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class);
|
||||
int quoted = jdbcOperations.queryForObject("select count(*) from names where name = 'Anna\"'", Integer.class);
|
||||
Assert.assertThat(result, is(3));
|
||||
Assert.assertThat(quoted, is(1));
|
||||
}
|
||||
}
|
||||
|
||||
@SpringBootApplication
|
||||
public static class PgcopySinkApplication {
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(PgcopySinkApplication.class, args);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
/*
|
||||
* Copyright 2017-2018 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.stream.app.pgcopy.sink;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.rules.ExpectedException;
|
||||
|
||||
import org.springframework.beans.factory.BeanCreationException;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.boot.test.util.TestPropertyValues;
|
||||
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
import static org.hamcrest.CoreMatchers.equalTo;
|
||||
import static org.hamcrest.MatcherAssert.assertThat;
|
||||
|
||||
/**
|
||||
* @author Thomas Risberg
|
||||
* @author Artem Bilan
|
||||
*/
|
||||
public class PgcopySinkPropertiesTests {
|
||||
|
||||
private AnnotationConfigApplicationContext context;
|
||||
|
||||
@Rule
|
||||
public ExpectedException thrown = ExpectedException.none();
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
this.context = new AnnotationConfigApplicationContext();
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() {
|
||||
this.context.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void tableNameIsRequired() {
|
||||
this.thrown.expect(BeanCreationException.class);
|
||||
this.thrown.expectMessage("Failed to bind properties under 'pgcopy' to org.springframework.cloud.stream.app.pgcopy.sink.PgcopySinkProperties");
|
||||
this.context.register(Conf.class);
|
||||
this.context.refresh();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void tableNameCanBeCustomized() {
|
||||
String table = "TEST_DATA";
|
||||
TestPropertyValues.of("pgcopy.table-name:" + table)
|
||||
.applyTo(this.context);
|
||||
this.context.register(Conf.class);
|
||||
this.context.refresh();
|
||||
PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class);
|
||||
assertThat(properties.getTableName(), equalTo(table));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void formatDefaultsToText() {
|
||||
TestPropertyValues.of("pgcopy.table-name: test")
|
||||
.applyTo(this.context);
|
||||
this.context.register(Conf.class);
|
||||
this.context.refresh();
|
||||
PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class);
|
||||
assertThat(properties.getFormat().toString(), equalTo("TEXT"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void formatCanBeCustomized() {
|
||||
String format = "CSV";
|
||||
TestPropertyValues.of("pgcopy.table-name: test", "pgcopy.format:" + format)
|
||||
.applyTo(this.context);
|
||||
this.context.register(Conf.class);
|
||||
this.context.refresh();
|
||||
PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class);
|
||||
assertThat(properties.getFormat().toString(), equalTo(format));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void nullCanBeCustomized() {
|
||||
String nullString = "@#$";
|
||||
TestPropertyValues.of("pgcopy.table-name: test",
|
||||
"pgcopy.format: CSV",
|
||||
"pgcopy.null-string: " + nullString)
|
||||
.applyTo(this.context);
|
||||
this.context.register(Conf.class);
|
||||
this.context.refresh();
|
||||
PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class);
|
||||
assertThat(properties.getNullString(), equalTo(nullString));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void delimiterCanBeCustomized() {
|
||||
String delimiter = "|";
|
||||
TestPropertyValues.of("pgcopy.table-name: test",
|
||||
"pgcopy.format: CSV",
|
||||
"pgcopy.delimiter: " + delimiter)
|
||||
.applyTo(this.context);
|
||||
this.context.register(Conf.class);
|
||||
this.context.refresh();
|
||||
PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class);
|
||||
assertThat(properties.getDelimiter(), equalTo(delimiter));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void quoteCanBeCustomized() {
|
||||
String quote = "'";
|
||||
TestPropertyValues.of("pgcopy.table-name: test",
|
||||
"pgcopy.format: CSV",
|
||||
"pgcopy.quote: " + quote)
|
||||
.applyTo(this.context);
|
||||
this.context.register(Conf.class);
|
||||
this.context.refresh();
|
||||
PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class);
|
||||
assertThat(String.valueOf(properties.getQuote()), equalTo(quote));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void escapeCanBeCustomized() {
|
||||
String escape = "~";
|
||||
TestPropertyValues.of("pgcopy.table-name: test",
|
||||
"pgcopy.format: CSV",
|
||||
"pgcopy.escape: " + escape)
|
||||
.applyTo(this.context);
|
||||
this.context.register(Conf.class);
|
||||
this.context.refresh();
|
||||
PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class);
|
||||
assertThat(String.valueOf(properties.getEscape()), equalTo(escape));
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@EnableConfigurationProperties(PgcopySinkProperties.class)
|
||||
static class Conf {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
/*
|
||||
* Copyright 2017-2018 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.stream.app.pgcopy.test;
|
||||
|
||||
import java.sql.Connection;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.jdbc.datasource.DataSourceUtils;
|
||||
|
||||
/**
|
||||
* JUnit {@link org.junit.Rule} that detects the fact that a PostgreSQL server is running on localhost.
|
||||
*
|
||||
* @author Thomas Risberg
|
||||
* @author Artem Bilan
|
||||
*/
|
||||
public class PostgresTestSupport extends AbstractExternalResourceTestSupport<DataSource> {
|
||||
|
||||
private ConfigurableApplicationContext context;
|
||||
|
||||
public PostgresTestSupport() {
|
||||
super("POSTGRES");
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void cleanupResource() {
|
||||
context.close();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void obtainResource() {
|
||||
context = new SpringApplicationBuilder(Config.class)
|
||||
.web(WebApplicationType.NONE)
|
||||
.run();
|
||||
DataSource dataSource = context.getBean(DataSource.class);
|
||||
Connection con = DataSourceUtils.getConnection(dataSource);
|
||||
DataSourceUtils.releaseConnection(con, dataSource);
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@EnableAutoConfiguration
|
||||
public static class Config {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
spring.datasource.url=jdbc:postgresql://127.0.0.1:5432/test
|
||||
spring.datasource.username=postgres
|
||||
spring.datasource.password=spring
|
||||
spring.datasource.driver-class-name=org.postgresql.Driver
|
||||
@@ -0,0 +1,2 @@
|
||||
drop table test_errors;
|
||||
create table test_errors (table_name varchar(255), error_message text, payload text);
|
||||
@@ -31,5 +31,6 @@
|
||||
<module>twitter-update-sink</module>
|
||||
<module>twitter-message-sink</module>
|
||||
<module>wavefront-sink</module>
|
||||
<module>pgcopy-sink</module>
|
||||
</modules>
|
||||
</project>
|
||||
|
||||
Reference in New Issue
Block a user