diff --git a/build.gradle b/build.gradle index 3b74c54a2c..f5940c99ad 100644 --- a/build.gradle +++ b/build.gradle @@ -337,6 +337,9 @@ project('spring-integration-file') { testCompile project(":spring-integration-redis") testCompile project(":spring-integration-redis").sourceSets.test.output testCompile project(":spring-integration-gemfire") + testCompile project(":spring-integration-jdbc") + testCompile "org.apache.derby:derby:$derbyVersion" + testCompile "org.apache.derby:derbyclient:$derbyVersion" testCompile "redis.clients:jedis:$jedisVersion" } } diff --git a/composite_filter_bugfix b/composite_filter_bugfix new file mode 100644 index 0000000000..e69de29bb2 diff --git a/spring-integration-file/src/test/java/int-derby.properties b/spring-integration-file/src/test/java/int-derby.properties new file mode 100644 index 0000000000..0ecac5970d --- /dev/null +++ b/spring-integration-file/src/test/java/int-derby.properties @@ -0,0 +1,4 @@ +# Placeholders for Derby: +int.drop.script=classpath:/org/springframework/integration/jdbc/schema-drop-derby.sql +int.schema.script=classpath:/org/springframework/integration/jdbc/schema-derby.sql +int.database.incrementer.class=org.springframework.jdbc.support.incrementer.DerbyMaxValueIncrementer diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests-context.xml new file mode 100644 index 0000000000..83887861fc --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests-context.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java index 3cf82cae61..589a04b9af 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java @@ -26,17 +26,24 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import javax.sql.DataSource; + import org.apache.geode.cache.CacheFactory; import org.junit.Test; +import org.junit.runner.RunWith; import org.mockito.Mockito; - +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.integration.gemfire.metadata.GemfireMetadataStore; +import org.springframework.integration.jdbc.metadata.JdbcMetadataStore; import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.integration.redis.metadata.RedisMetadataStore; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Gary Russell @@ -44,8 +51,14 @@ import org.springframework.integration.redis.rules.RedisAvailableTests; * @since 4.0 * */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext // close at the end after class public class PersistentAcceptOnceFileListFilterExternalStoreTests extends RedisAvailableTests { + @Autowired + private DataSource dataSource; + @Test @RedisAvailable public void testFileSystemWithRedisMetadataStore() throws Exception { @@ -69,6 +82,13 @@ public class PersistentAcceptOnceFileListFilterExternalStoreTests extends RedisA this.testFileSystem(new GemfireMetadataStore(new CacheFactory().create())); } + @Test + public void testFileSystemWithJdbcMetadataStore() throws Exception { + JdbcMetadataStore metadataStore = new JdbcMetadataStore(dataSource); + metadataStore.afterPropertiesSet(); + this.testFileSystem(metadataStore); + } + private void testFileSystem(ConcurrentMetadataStore store) throws Exception { final AtomicBoolean suspend = new AtomicBoolean(); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStore.java new file mode 100644 index 0000000000..ab5dfd35b3 --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStore.java @@ -0,0 +1,223 @@ +/* + * Copyright 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 + * + * http://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.integration.jdbc.metadata; + +import javax.sql.DataSource; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.dao.EmptyResultDataAccessException; +import org.springframework.integration.metadata.ConcurrentMetadataStore; +import org.springframework.integration.metadata.MetadataStore; +import org.springframework.jdbc.core.JdbcOperations; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.util.Assert; + +/** + * Implementation of {@link MetadataStore} using a relational database via JDBC. SQL scripts to create the necessary + * tables are packaged as org/springframework/integration/jdbc/schema-*.sql, where * is the + * target database type. + * + * @author Bojan Vukasovic + * @since 5.0 + */ +public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingBean { + + /** + * Default value for the table prefix property. + */ + public static final String DEFAULT_TABLE_PREFIX = "INT_"; + + private volatile String tablePrefix = DEFAULT_TABLE_PREFIX; + + private volatile String region = "DEFAULT"; + + private String getValueQuery = "SELECT METADATA_VALUE FROM %SMETADATA_STORE WHERE METADATA_KEY=? AND REGION=?"; + + private String getValueForUpdateQuery = "SELECT METADATA_VALUE FROM %SMETADATA_STORE WHERE METADATA_KEY=? AND REGION=? FOR UPDATE"; + + private String replaceValueQuery = "UPDATE %SMETADATA_STORE SET METADATA_VALUE=? WHERE METADATA_KEY=? AND METADATA_VALUE=? AND REGION=?"; + + private String replaceValueByKeyQuery = "UPDATE %SMETADATA_STORE SET METADATA_VALUE=? WHERE METADATA_KEY=? AND REGION=?"; + + private String removeValueQuery = "DELETE FROM %SMETADATA_STORE WHERE METADATA_KEY=? AND REGION=?"; + + private String putIfAbsentValueQuery = "INSERT INTO %SMETADATA_STORE(METADATA_KEY, METADATA_VALUE, REGION) " + + "SELECT ?, ?, ? FROM %SMETADATA_STORE WHERE METADATA_KEY=? AND REGION=? HAVING COUNT(*)=0"; + + private final JdbcOperations jdbcTemplate; + + @Override + public void afterPropertiesSet() throws Exception { + this.getValueQuery = String.format(this.getValueQuery, this.tablePrefix); + this.getValueForUpdateQuery = String.format(this.getValueForUpdateQuery, this.tablePrefix); + this.replaceValueQuery = String.format(this.replaceValueQuery, this.tablePrefix); + this.replaceValueByKeyQuery = String.format(this.replaceValueByKeyQuery, this.tablePrefix); + this.removeValueQuery = String.format(this.removeValueQuery, this.tablePrefix); + this.putIfAbsentValueQuery = String.format(this.putIfAbsentValueQuery, this.tablePrefix, this.tablePrefix); + } + + /** + * Instantiate a {@link JdbcMetadataStore} using provided dataSource {@link DataSource}. + * @param dataSource a {@link DataSource} + */ + public JdbcMetadataStore(DataSource dataSource) { + this(new JdbcTemplate(dataSource)); + } + + /** + * Instantiate a {@link JdbcMetadataStore} using provided jdbcOperations {@link JdbcOperations}. + * @param jdbcOperations a {@link JdbcOperations} + */ + public JdbcMetadataStore(JdbcOperations jdbcOperations) { + Assert.notNull(jdbcOperations, "'jdbcOperations' must not be null"); + this.jdbcTemplate = jdbcOperations; + } + + /** + * Public setter for the table prefix property. This will be prefixed to all the table names before queries are + * executed. Defaults to {@link #DEFAULT_TABLE_PREFIX}. + * + * @param tablePrefix the tablePrefix to set + */ + public void setTablePrefix(String tablePrefix) { + this.tablePrefix = tablePrefix; + } + + /** + * A unique grouping identifier for all messages persisted with this store. Using multiple regions allows the store + * to be partitioned (if necessary) for different purposes. Defaults to DEFAULT. + * + * @param region the region name to set + */ + public void setRegion(String region) { + Assert.hasText(region, "Region must not be null or empty."); + this.region = region; + } + + @Override + @Transactional + public String putIfAbsent(String key, String value) { + Assert.notNull(key, "'key' cannot be null"); + Assert.notNull(value, "'value' cannot be null"); + while (true) { + //try to insert if does not exists + int affectedRows = tryToPutIfAbsent(key, value); + if (affectedRows > 0) { + //it was not in the table, so we have just inserted it + return null; + } + else { + //value should be in table. try to return it + try { + return this.jdbcTemplate.queryForObject(this.getValueQuery, String.class, key, this.region); + } + catch (EmptyResultDataAccessException e) { + //somebody deleted it between calls. try to insert again (go to beginning of while loop) + } + } + } + } + + private int tryToPutIfAbsent(String key, String value) { + return this.jdbcTemplate.update(this.putIfAbsentValueQuery, ps -> { + ps.setString(1, key); + ps.setString(2, value); + ps.setString(3, this.region); + ps.setString(4, key); + ps.setString(5, this.region); + }); + } + + @Override + @Transactional + public boolean replace(String key, String oldValue, String newValue) { + Assert.notNull(key, "'key' cannot be null"); + Assert.notNull(oldValue, "'oldValue' cannot be null"); + Assert.notNull(newValue, "'newValue' cannot be null"); + int affectedRows = this.jdbcTemplate.update(this.replaceValueQuery, ps -> { + ps.setString(1, newValue); + ps.setString(2, key); + ps.setString(3, oldValue); + ps.setString(4, this.region); + }); + return affectedRows > 0; + } + + @Override + @Transactional + public void put(String key, String value) { + Assert.notNull(key, "'key' cannot be null"); + Assert.notNull(value, "'value' cannot be null"); + while (true) { + //try to insert if does not exist, if exists we will try to update it + int affectedRows = tryToPutIfAbsent(key, value); + if (affectedRows == 0) { + //since value is not inserted, means it is already present + try { + //lock row for updating + this.jdbcTemplate.queryForObject(this.getValueForUpdateQuery, String.class, key, this.region); + } + catch (EmptyResultDataAccessException e) { + //if there are no rows with this key, somebody deleted it in between two calls + continue; //try to insert again from beginning + } + //lock successful, so - replace + this.jdbcTemplate.update(this.replaceValueByKeyQuery, ps -> { + ps.setString(1, value); + ps.setString(2, key); + ps.setString(3, this.region); + }); + } + return; + } + } + + @Override + @Transactional + public String get(String key) { + Assert.notNull(key, "'key' cannot be null"); + try { + return this.jdbcTemplate.queryForObject(this.getValueQuery, String.class, key, this.region); + } + catch (EmptyResultDataAccessException e) { + //if there are no rows with this key, return null + return null; + } + } + + @Override + @Transactional + public String remove(String key) { + Assert.notNull(key, "'key' cannot be null"); + String oldValue; + try { + //select old value and lock row for removal + oldValue = this.jdbcTemplate.queryForObject(this.getValueForUpdateQuery, String.class, key, this.region); + } + catch (EmptyResultDataAccessException e) { + //key is not present, so no need to delete it + return null; + } + //delete row and return old value + int updated = this.jdbcTemplate.update(this.removeValueQuery, key, this.region); + if (updated != 0) { + return oldValue; + } + return null; + } +} diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/metadata/package-info.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/metadata/package-info.java new file mode 100644 index 0000000000..9c34c86fc1 --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/metadata/package-info.java @@ -0,0 +1,4 @@ +/** + * Contains JDBC implementation of MetadataStore + */ +package org.springframework.integration.jdbc.metadata; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql index f7df8fb7b6..477d808db6 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql @@ -51,3 +51,11 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); + + +CREATE TABLE INT_METADATA_STORE ( + METADATA_KEY VARCHAR(255) NOT NULL, + METADATA_VALUE VARCHAR(4000), + REGION VARCHAR(100) NOT NULL, + constraint METADATA_STORE primary key (METADATA_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql index ccc16c1dfc..0a69f37da4 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql @@ -51,3 +51,11 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); + + +CREATE TABLE INT_METADATA_STORE ( + METADATA_KEY VARCHAR(255) NOT NULL, + METADATA_VALUE VARCHAR(4000), + REGION VARCHAR(100) NOT NULL, + constraint METADATA_STORE primary key (METADATA_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql index 6805ae91d3..a548cca409 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql @@ -8,5 +8,5 @@ DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; DROP TABLE INT_LOCK ; DROP TABLE INT_CHANNEL_MESSAGE ; +DROP TABLE INT_METADATA_STORE ; DROP SEQUENCE INT_MESSAGE_SEQ ; - diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql index d49ada18b6..ad8d79d1af 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql @@ -8,4 +8,4 @@ DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; DROP TABLE INT_LOCK ; DROP TABLE INT_CHANNEL_MESSAGE ; - +DROP TABLE INT_METADATA_STORE ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql index 8725447fe2..77e648a444 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql @@ -8,5 +8,5 @@ DROP TABLE INT_MESSAGE_GROUP IF EXISTS; DROP TABLE INT_GROUP_TO_MESSAGE IF EXISTS; DROP TABLE INT_LOCK IF EXISTS; DROP TABLE INT_CHANNEL_MESSAGE IF EXISTS; +DROP TABLE INT_METADATA_STORE IF EXISTS; DROP SEQUENCE INT_MESSAGE_SEQ IF EXISTS; - diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql index 8725447fe2..77e648a444 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql @@ -8,5 +8,5 @@ DROP TABLE INT_MESSAGE_GROUP IF EXISTS; DROP TABLE INT_GROUP_TO_MESSAGE IF EXISTS; DROP TABLE INT_LOCK IF EXISTS; DROP TABLE INT_CHANNEL_MESSAGE IF EXISTS; +DROP TABLE INT_METADATA_STORE IF EXISTS; DROP SEQUENCE INT_MESSAGE_SEQ IF EXISTS; - diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql index 35858e3de6..aa45c554ca 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql @@ -5,4 +5,4 @@ DROP TABLE IF EXISTS INT_MESSAGE_GROUP ; DROP TABLE IF EXISTS INT_GROUP_TO_MESSAGE ; DROP TABLE IF EXISTS INT_LOCK ; DROP TABLE IF EXISTS INT_CHANNEL_MESSAGE ; - +DROP TABLE IF EXISTS INT_METADATA_STORE ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle.sql index 6805ae91d3..a548cca409 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle.sql @@ -8,5 +8,5 @@ DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; DROP TABLE INT_LOCK ; DROP TABLE INT_CHANNEL_MESSAGE ; +DROP TABLE INT_METADATA_STORE ; DROP SEQUENCE INT_MESSAGE_SEQ ; - diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql index 6805ae91d3..a548cca409 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql @@ -8,5 +8,5 @@ DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; DROP TABLE INT_LOCK ; DROP TABLE INT_CHANNEL_MESSAGE ; +DROP TABLE INT_METADATA_STORE ; DROP SEQUENCE INT_MESSAGE_SEQ ; - diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql index 6805ae91d3..a548cca409 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql @@ -8,5 +8,5 @@ DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; DROP TABLE INT_LOCK ; DROP TABLE INT_CHANNEL_MESSAGE ; +DROP TABLE INT_METADATA_STORE ; DROP SEQUENCE INT_MESSAGE_SEQ ; - diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql index 6805ae91d3..a548cca409 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql @@ -8,5 +8,5 @@ DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; DROP TABLE INT_LOCK ; DROP TABLE INT_CHANNEL_MESSAGE ; +DROP TABLE INT_METADATA_STORE ; DROP SEQUENCE INT_MESSAGE_SEQ ; - diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql index 298dbcabbc..bbe50da7b3 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql @@ -51,3 +51,11 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); + + +CREATE TABLE INT_METADATA_STORE ( + METADATA_KEY VARCHAR(255) NOT NULL, + METADATA_VALUE VARCHAR(4000), + REGION VARCHAR(100) NOT NULL, + constraint METADATA_STORE primary key (METADATA_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql index 7b74fbe416..1b9f443709 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql @@ -51,3 +51,11 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); + + +CREATE TABLE INT_METADATA_STORE ( + METADATA_KEY VARCHAR(255) NOT NULL, + METADATA_VALUE VARCHAR(4000), + REGION VARCHAR(100) NOT NULL, + constraint METADATA_STORE primary key (METADATA_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql index 123fd295f1..a6f3587375 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql @@ -51,3 +51,11 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); + + +CREATE TABLE INT_METADATA_STORE ( + METADATA_KEY VARCHAR(255) NOT NULL, + METADATA_VALUE VARCHAR(4000), + REGION VARCHAR(100) NOT NULL, + constraint METADATA_STORE primary key (METADATA_KEY, REGION) +) ENGINE=InnoDB; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql index 9eeef49c25..eda21929a7 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql @@ -51,3 +51,11 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); + + +CREATE TABLE INT_METADATA_STORE ( + METADATA_KEY VARCHAR2(255) NOT NULL, + METADATA_VALUE VARCHAR2(4000), + REGION VARCHAR2(100) NOT NULL, + constraint METADATA_STORE primary key (METADATA_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql index 97d4f69c56..c11e03cb71 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql @@ -51,3 +51,11 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); + + +CREATE TABLE INT_METADATA_STORE ( + METADATA_KEY VARCHAR(255) NOT NULL, + METADATA_VALUE VARCHAR(4000), + REGION VARCHAR(100) NOT NULL, + constraint METADATA_STORE primary key (METADATA_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql index 7741ef3e15..6d5bfff80e 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql @@ -51,3 +51,11 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); + + +CREATE TABLE INT_METADATA_STORE ( + METADATA_KEY VARCHAR(255) NOT NULL, + METADATA_VALUE VARCHAR(4000), + REGION VARCHAR(100) NOT NULL, + constraint METADATA_STORE primary key (METADATA_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql index 65f2048c54..5863509f79 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql @@ -51,3 +51,11 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); + + +CREATE TABLE INT_METADATA_STORE ( + METADATA_KEY VARCHAR(255) NOT NULL, + METADATA_VALUE VARCHAR(4000), + REGION VARCHAR(100) NOT NULL, + constraint METADATA_STORE primary key (METADATA_KEY, REGION) +) LOCK DATAROWS; diff --git a/spring-integration-jdbc/src/main/sql/destroy.sql.vpp b/spring-integration-jdbc/src/main/sql/destroy.sql.vpp index 8fa27afe22..d32c506917 100644 --- a/spring-integration-jdbc/src/main/sql/destroy.sql.vpp +++ b/spring-integration-jdbc/src/main/sql/destroy.sql.vpp @@ -12,7 +12,7 @@ DROP TABLE $!{IFEXISTSBEFORE} INT_MESSAGE_GROUP $!{IFEXISTS}; DROP TABLE $!{IFEXISTSBEFORE} INT_GROUP_TO_MESSAGE $!{IFEXISTS}; DROP TABLE $!{IFEXISTSBEFORE} INT_LOCK $!{IFEXISTS}; DROP TABLE $!{IFEXISTSBEFORE} INT_CHANNEL_MESSAGE $!{IFEXISTS}; +DROP TABLE $!{IFEXISTSBEFORE} INT_METADATA_STORE $!{IFEXISTS}; #if(${INT_MESSAGE_SEQ}) DROP SEQUENCE INT_MESSAGE_SEQ $!{IFEXISTS}; #end - diff --git a/spring-integration-jdbc/src/main/sql/schema.sql.vpp b/spring-integration-jdbc/src/main/sql/schema.sql.vpp index 9707e3855c..950a790bf2 100644 --- a/spring-integration-jdbc/src/main/sql/schema.sql.vpp +++ b/spring-integration-jdbc/src/main/sql/schema.sql.vpp @@ -52,3 +52,11 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); + + +CREATE TABLE INT_METADATA_STORE ( + METADATA_KEY ${VARCHAR}(255) NOT NULL, + METADATA_VALUE ${VARCHAR}(4000), + REGION ${VARCHAR}(100) NOT NULL, + constraint METADATA_STORE primary key (METADATA_KEY, REGION) +)#if(${VOODOO}) ${VOODOO}#end; diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStoreTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStoreTests-context.xml new file mode 100644 index 0000000000..83887861fc --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStoreTests-context.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStoreTests.java new file mode 100644 index 0000000000..8556a3375f --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStoreTests.java @@ -0,0 +1,108 @@ +/* + * Copyright 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 + * + * http://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.integration.jdbc.metadata; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +import javax.sql.DataSource; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.transaction.annotation.Transactional; + +/** + * @author Bojan Vukasovic + * @since 5.0 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext // close at the end after class +@Transactional +public class JdbcMetadataStoreTests { + + @Autowired + private DataSource dataSource; + + + private JdbcMetadataStore metadataStore; + + @Before + public void init() throws Exception { + metadataStore = new JdbcMetadataStore(dataSource); + metadataStore.afterPropertiesSet(); + } + + @Test + public void keyAndValuesArePreservedOnPut() { + metadataStore.put("foo", "bar"); + metadataStore.put("foo", "bar1"); + metadataStore.put("foo2", "bar2"); + String bar1 = metadataStore.get("foo"); + String bar2 = metadataStore.get("foo2"); + assertEquals("bar1", bar1); + assertEquals("bar2", bar2); + } + + @Test + public void keyAndValuesAreNotPreservedOnRemove() { + metadataStore.put("foo", "bar"); + metadataStore.put("foo2", "bar2"); + metadataStore.remove("foo"); + String bar = metadataStore.get("foo"); + metadataStore.remove("foo2"); + String bar2 = metadataStore.get("foo2"); + assertNull(bar); + assertNull(bar2); + } + + @Test + public void keyAndValuesAreNotOverwrittenOnPutIfAbsent() { + metadataStore.put("foo", "bar"); + metadataStore.putIfAbsent("foo", "bar1"); + String bar = metadataStore.get("foo"); + assertEquals("bar", bar); + } + + @Test + public void nonExistentKeyIsNotRemoved() { + metadataStore.remove("non-existent"); + String ne = metadataStore.get("non-existent"); + assertNull(ne); + } + + @Test + public void existingKeyValueIsReplacedWithNewValueWhenOldValueMatches() { + metadataStore.put("foo", "bar"); + metadataStore.replace("foo", "bar", "bar2"); + String bar2 = metadataStore.get("foo"); + assertEquals("bar2", bar2); + } + + @Test + public void existingKeyValueIsNotReplacedWithNewValueWhenOldValueDoesNotMatch() { + metadataStore.put("foo", "bar"); + metadataStore.replace("foo", "bar1", "bar2"); + String bar = metadataStore.get("foo"); + assertEquals("bar", bar); + } +}