Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3339: Add priority to the JdbcChannelMS

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3339

INT-3339: Improve `priority` logic

INT-3339: Add `MESSAGE_SEQUENCE` stuff

INT-3339: Docs

INT-3339: Polishing and PR comments

INT-3339 Polishing

Fix Oracle Test Case.
Doc polishing.
This commit is contained in:
Artem Bilan
2014-04-07 16:02:33 +03:00
committed by Gary Russell
parent 3f40c40641
commit 64a7a7e021
41 changed files with 519 additions and 429 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/src/main/java/org/springframework/integration/store/BasicMessageGroupStore.java
View File

@@ -63,4 +63,10 @@ public interface BasicMessageGroupStore {
*/
Message<?> pollMessageFromGroup(Object groupId);
/**
* Remove the message group with this id.
*
* @param groupId The id of the group to remove.
*/
void removeMessageGroup(Object groupId);
}

7
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java
View File

@@ -59,13 +59,6 @@ public interface MessageGroupStore extends BasicMessageGroupStore {
*/
MessageGroup removeMessageFromGroup(Object key, Message<?> messageToRemove);
/**
* Remove the message group with this id.
*
* @param groupId The id of the group to remove.
*/
void removeMessageGroup(Object groupId);
/**
* Register a callback for when a message group is expired through {@link #expireMessageGroups(long)}.
*

1
spring-integration-jdbc/.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
src/test/java/org/springframework/integration/jdbc/store/channel/DataSource-oracle-context-*

285
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java
View File

@@ -18,7 +18,6 @@ import java.sql.SQLException;
import java.sql.Types;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -32,12 +31,17 @@ import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.core.serializer.support.DeserializingConverter;
import org.springframework.core.serializer.support.SerializingConverter;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.jdbc.JdbcMessageStore;
import org.springframework.integration.jdbc.store.channel.ChannelMessageStoreQueryProvider;
import org.springframework.integration.jdbc.store.channel.DerbyChannelMessageStoreQueryProvider;
@@ -45,12 +49,13 @@ import org.springframework.integration.jdbc.store.channel.MessageRowMapper;
import org.springframework.integration.jdbc.store.channel.MySqlChannelMessageStoreQueryProvider;
import org.springframework.integration.jdbc.store.channel.OracleChannelMessageStoreQueryProvider;
import org.springframework.integration.jdbc.store.channel.PostgresChannelMessageStoreQueryProvider;
import org.springframework.integration.store.AbstractMessageGroupStore;
import org.springframework.integration.store.ChannelMessageStore;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.PriorityCapableChannelMessageStore;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.transaction.TransactionSynchronizationFactory;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcOperations;
@@ -90,8 +95,7 @@ import org.springframework.util.StringUtils;
* @since 2.2
*/
@ManagedResource
public class JdbcChannelMessageStore extends AbstractMessageGroupStore
implements InitializingBean, ChannelMessageStore {
public class JdbcChannelMessageStore implements PriorityCapableChannelMessageStore, InitializingBean, BeanFactoryAware {
private static final Log logger = LogFactory.getLog(JdbcChannelMessageStore.class);
@@ -117,8 +121,6 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
private ChannelMessageStoreQueryProvider channelMessageStoreQueryProvider;
public static final int DEFAULT_LONG_STRING_LENGTH = 2500;
/**
* The name of the message header that stores a flag to indicate that the message has been saved. This is an
* optimization for the put method.
@@ -130,6 +132,8 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
*/
public static final String CREATED_DATE_KEY = JdbcChannelMessageStore.class.getSimpleName() + ".CREATED_DATE";
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile String region = DEFAULT_REGION;
private volatile String tablePrefix = DEFAULT_TABLE_PREFIX;
@@ -148,6 +152,8 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
private boolean usingIdCache = false;
private boolean priorityEnabled;
/**
* Convenient constructor for configuration use.
*/
@@ -171,8 +177,6 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
this.jdbcTemplate.setFetchSize(1);
this.jdbcTemplate.setMaxRows(1);
this.jdbcTemplate.afterPropertiesSet();
}
/**
@@ -188,8 +192,6 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
this.jdbcTemplate.setFetchSize(1);
this.jdbcTemplate.setMaxRows(1);
this.jdbcTemplate.afterPropertiesSet();
}
/**
@@ -197,7 +199,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
*
* @param deserializer the deserializer to set
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
@SuppressWarnings({"unchecked", "rawtypes"})
public void setDeserializer(Deserializer<? extends Message<?>> deserializer) {
this.deserializer = new DeserializingConverter((Deserializer) deserializer);
}
@@ -217,15 +219,6 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
this.jdbcTemplate = jdbcTemplate;
}
/**
* Method not implemented.
* @throws UnsupportedOperationException Method not supported.
*/
@Override
public void setLastReleasedSequenceNumberForGroup(Object groupId, final int sequenceNumber) {
throw new UnsupportedOperationException("Not implemented");
}
/**
* Override the {@link LobHandler} that is used to create and unpack large objects in SQL queries. The default is
* fine for almost all platforms, but some Oracle drivers require a native implementation.
@@ -355,6 +348,20 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
this.usingIdCache = usingIdCache;
}
public void setPriorityEnabled(boolean priorityEnabled) {
this.priorityEnabled = priorityEnabled;
}
@Override
public boolean isPriorityEnabled() {
return this.priorityEnabled;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
/**
* Check mandatory properties ({@link DataSource} and
* {@link #setChannelMessageStoreQueryProvider(ChannelMessageStoreQueryProvider)}). If no {@link MessageRowMapper} was
@@ -379,30 +386,31 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
}
if (this.jdbcTemplate.getFetchSize() != 1 && logger.isWarnEnabled()) {
logger.warn("The jdbcTemplate's fetchsize is not 1 but %s. This may cause FIFO issues with Oracle databases.");
logger.warn("The jdbcTemplate's fetchsize is not 1. This may cause FIFO issues with Oracle databases.");
}
this.jdbcTemplate.afterPropertiesSet();
}
/**
* Store a message in the database. The groupId identifies the channel for which
* the message is to be stored.
*
* Keep in mind that the actual groupdId (Channel
* Keep in mind that the actual groupId (Channel
* Identifier) is converted to a String-based UUID identifier.
*
* @param groupId the group id to store the message under
* @param message a message
*/
@Override
@SuppressWarnings({ "rawtypes", "unchecked" })
public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
@SuppressWarnings({"rawtypes", "unchecked"})
public MessageGroup addMessageToGroup(Object groupId, final Message<?> message) {
final String groupKey = getKey(groupId);
final long createdDate = System.currentTimeMillis();
final Message<?> result = this.getMessageBuilderFactory().fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE)
.setHeader(CREATED_DATE_KEY, new Long(createdDate)).build();
final Message<?> result = this.messageBuilderFactory.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE)
.setHeader(CREATED_DATE_KEY, createdDate).build();
final Map innerMap = (Map) new DirectFieldAccessor(result.getHeaders()).getPropertyValue("headers");
// using reflection to set ID since it is immutable through MessageHeaders
@@ -414,90 +422,30 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
jdbcTemplate.update(getQuery(channelMessageStoreQueryProvider.getCreateMessageQuery()), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()){
if (logger.isDebugEnabled()) {
logger.debug("Inserting message with id key=" + messageId);
}
ps.setString(1, messageId);
ps.setString(2, groupKey);
ps.setString(3, region);
ps.setLong(4, createdDate);
lobHandler.getLobCreator().setBlobAsBytes(ps, 5, messageBytes);
Integer priority = new IntegrationMessageHeaderAccessor(message).getPriority();
if (JdbcChannelMessageStore.this.priorityEnabled && priority != null) {
ps.setInt(5, priority);
}
else {
ps.setNull(5, Types.NUMERIC);
}
lobHandler.getLobCreator().setBlobAsBytes(ps, 6, messageBytes);
}
});
return getMessageGroup(groupId);
}
/**
* Method not implemented.
*
* @throws UnsupportedOperationException Method not supported.
*/
@Override
public void completeGroup(Object groupId) {
throw new UnsupportedOperationException("Not implemented");
}
/**
* This method executes a call to the DB to get the oldest Message in the
* MessageGroup which in the context of the {@link JdbcChannelMessageStore}
* means the channel identifier.
*
* @param groupIdKey String representation of message group (Channel) ID
* @return a message; could be null if query produced no Messages
*/
protected Message<?> doPollForMessage(String groupIdKey) {
final NamedParameterJdbcTemplate namedParameterJdbcTemplate = new NamedParameterJdbcTemplate(jdbcTemplate);
final MapSqlParameterSource parameters = new MapSqlParameterSource();
parameters.addValue("region", region);
parameters.addValue("group_key", groupIdKey);
final String query;
final List<Message<?>> messages;
this.idCacheReadLock.lock();
try {
if (this.usingIdCache && !this.idCache.isEmpty()) {
query = getQuery(this.channelMessageStoreQueryProvider.getPollFromGroupExcludeIdsQuery());
parameters.addValue("message_ids", idCache);
} else {
query = getQuery(this.channelMessageStoreQueryProvider.getPollFromGroupQuery());
}
messages = namedParameterJdbcTemplate.query(query, parameters, messageRowMapper);
}
finally {
this.idCacheReadLock.unlock();
}
Assert.isTrue(messages.size() == 0 || messages.size() == 1);
if (messages.size() > 0){
final Message<?>message = messages.get(0);
final String messageId = message.getHeaders().getId().toString();
if (this.usingIdCache) {
this.idCacheWriteLock.lock();
try {
boolean added = this.idCache.add(messageId);
if (logger.isDebugEnabled()) {
logger.debug(String.format("Polled message with id '%s' added: '%s'.", messageId, added));
}
}
finally {
this.idCacheWriteLock.unlock();
}
}
return message;
}
return null;
}
/**
* Helper method that converts the channel id to a UUID using
* {@link UUIDConverter#getUUID(Object)}.
@@ -509,27 +457,6 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
return input == null ? null : UUIDConverter.getUUID(input).toString();
}
/**
* Method not implemented.
* @return The message count.
* @throws UnsupportedOperationException Method not supported.
*/
@ManagedAttribute
public long getMessageCount() {
throw new UnsupportedOperationException("Not implemented");
}
/**
* Method not implemented.
* @return The message count.
* @throws UnsupportedOperationException Method not supported.
*/
@Override
@ManagedAttribute
public int getMessageCountForAllMessageGroups() {
throw new UnsupportedOperationException("Not implemented");
}
/**
* Not fully used. Only wraps the provided group id.
*/
@@ -544,10 +471,10 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
* @return The message group count.
* @throws UnsupportedOperationException Method not supported.
*/
@Override
@ManagedAttribute
public int getMessageGroupCount() {
throw new UnsupportedOperationException("Not implemented");
return this.jdbcTemplate.queryForObject(this.getQuery("SELECT COUNT(DISTINCT GROUP_KEY) from %PREFIX%CHANNEL_MESSAGE where REGION = ?"),
Integer.class, this.region);
}
/**
@@ -569,16 +496,6 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
return query;
}
/**
* Method not implemented.
*
* @throws UnsupportedOperationException Method not supported.
*/
@Override
public Iterator<MessageGroup> iterator() {
throw new UnsupportedOperationException("Not implemented");
}
/**
* Returns the number of messages persisted for the specified channel id (groupId)
* and the specified region ({@link #setRegion(String)}).
@@ -593,6 +510,11 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
Integer.class, key, this.region);
}
public void removeMessageGroup(Object groupId) {
this.jdbcTemplate.update(this.getQuery(this.channelMessageStoreQueryProvider.getDeleteMessageGroupQuery()),
this.getKey(groupId), this.region);
}
/**
* Polls the database for a new message that is persisted for the given
* group id which represents the channel identifier.
@@ -603,7 +525,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
final String key = getKey(groupId);
final Message<?> polledMessage = this.doPollForMessage(key);
if (polledMessage != null){
if (polledMessage != null) {
if (!this.doRemoveMessageFromGroup(groupId, polledMessage)) {
return null;
}
@@ -613,25 +535,81 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
}
/**
* Remove a single message from the database.
*
* @param groupId The channel id to remove the message from.
* @param messageToRemove The message to remove.
* This method executes a call to the DB to get the oldest Message in the
* MessageGroup which in the context of the {@link JdbcChannelMessageStore}
* means the channel identifier.
*
* @param groupIdKey String representation of message group (Channel) ID
* @return a message; could be null if query produced no Messages
*/
@Override
public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
protected Message<?> doPollForMessage(String groupIdKey) {
this.doRemoveMessageFromGroup(groupId, messageToRemove);
final NamedParameterJdbcTemplate namedParameterJdbcTemplate = new NamedParameterJdbcTemplate(jdbcTemplate);
final MapSqlParameterSource parameters = new MapSqlParameterSource();
return getMessageGroup(groupId);
parameters.addValue("region", region);
parameters.addValue("group_key", groupIdKey);
String query;
final List<Message<?>> messages;
this.idCacheReadLock.lock();
try {
if (this.usingIdCache && !this.idCache.isEmpty()) {
if (this.priorityEnabled) {
query = getQuery(this.channelMessageStoreQueryProvider.getPriorityPollFromGroupExcludeIdsQuery());
}
else {
query = getQuery(this.channelMessageStoreQueryProvider.getPollFromGroupExcludeIdsQuery());
}
parameters.addValue("message_ids", idCache);
}
else {
if (this.priorityEnabled) {
query = getQuery(this.channelMessageStoreQueryProvider.getPriorityPollFromGroupQuery());
}
else {
query = getQuery(this.channelMessageStoreQueryProvider.getPollFromGroupQuery());
}
}
messages = namedParameterJdbcTemplate.query(query, parameters, messageRowMapper);
}
finally {
this.idCacheReadLock.unlock();
}
Assert.isTrue(messages.size() == 0 || messages.size() == 1);
if (messages.size() > 0) {
final Message<?> message = messages.get(0);
final String messageId = message.getHeaders().getId().toString();
if (this.usingIdCache) {
this.idCacheWriteLock.lock();
try {
boolean added = this.idCache.add(messageId);
if (logger.isDebugEnabled()) {
logger.debug(String.format("Polled message with id '%s' added: '%s'.", messageId, added));
}
}
finally {
this.idCacheWriteLock.unlock();
}
}
return message;
}
return null;
}
private boolean doRemoveMessageFromGroup(Object groupId, Message<?> messageToRemove) {
final UUID id = messageToRemove.getHeaders().getId();
int updated = jdbcTemplate.update(getQuery(channelMessageStoreQueryProvider.getDeleteMessageQuery()), new Object[] { getKey(id), getKey(groupId), region }, new int[] {
Types.VARCHAR, Types.VARCHAR, Types.VARCHAR });
int updated = jdbcTemplate.update(getQuery(channelMessageStoreQueryProvider.getDeleteMessageQuery()),
new Object[] {getKey(id), getKey(groupId), region}, new int[] {Types.VARCHAR, Types.VARCHAR, Types.VARCHAR});
boolean result = updated != 0;
if (result) {
@@ -678,25 +656,4 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore
return this.idCache.size();
}
/**
* Will remove all messages from the message channel.
*/
@Override
public void removeMessageGroup(Object groupId) {
final String groupKey = getKey(groupId);
jdbcTemplate.update(getQuery(channelMessageStoreQueryProvider.getDeleteMessageGroupQuery()), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()){
logger.debug("Marking messages with group key=" + groupKey);
}
ps.setString(1, groupKey);
ps.setString(2, region);
}
});
}
}

10
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/AbstractChannelMessageStoreQueryProvider.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -15,6 +15,7 @@ package org.springframework.integration.jdbc.store.channel;
/**
* @author Gunnar Hillert
* @author Artem Bilan
* @since 2.2
*/
public abstract class AbstractChannelMessageStoreQueryProvider implements ChannelMessageStoreQueryProvider {
@@ -23,9 +24,6 @@ public abstract class AbstractChannelMessageStoreQueryProvider implements Channe
return "SELECT COUNT(MESSAGE_ID) from %PREFIX%CHANNEL_MESSAGE where GROUP_KEY=? and REGION=?";
}
public abstract String getPollFromGroupExcludeIdsQuery();
public abstract String getPollFromGroupQuery();
public String getMessageQuery() {
return "SELECT MESSAGE_ID, CREATED_DATE, MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE where MESSAGE_ID=? and GROUP_KEY=? and REGION=?";
}
@@ -39,8 +37,8 @@ public abstract class AbstractChannelMessageStoreQueryProvider implements Channe
}
public String getCreateMessageQuery() {
return "INSERT into %PREFIX%CHANNEL_MESSAGE(MESSAGE_ID, GROUP_KEY, REGION, CREATED_DATE, MESSAGE_BYTES)"
+ " values (?, ?, ?, ?, ?)";
return "INSERT into %PREFIX%CHANNEL_MESSAGE(MESSAGE_ID, GROUP_KEY, REGION, CREATED_DATE, MESSAGE_PRIORITY, MESSAGE_BYTES)"
+ " values (?, ?, ?, ?, ?, ?)";
}
public String getDeleteMessageGroupQuery() {

18
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/ChannelMessageStoreQueryProvider.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -20,6 +20,7 @@ import org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
* {@link JdbcChannelMessageStore} to provide database-specific queries.
*
* @author Gunnar Hillert
* @author Artem Bilan
* @since 2.2
*/
public interface ChannelMessageStoreQueryProvider {
@@ -47,6 +48,21 @@ public interface ChannelMessageStoreQueryProvider {
*/
String getPollFromGroupQuery();
/**
* Get the query used to retrieve the oldest message by priority for a channel excluding
* messages that match the provided message ids.
*
* @return Sql Query
*/
String getPriorityPollFromGroupExcludeIdsQuery();
/**
* Get the query used to retrieve the oldest message by priority for a channel.
*
* @return Sql Query
*/
String getPriorityPollFromGroupQuery();
/**
* Query that retrieves a message for the provided message id, channel and
* region.

22
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/DerbyChannelMessageStoreQueryProvider.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -14,6 +14,7 @@ package org.springframework.integration.jdbc.store.channel;
/**
* @author Gunnar Hillert
* @author Artem Bilan
* @since 2.2
*
* https://blogs.oracle.com/kah/entry/derby_10_5_preview_fetch
@@ -24,14 +25,29 @@ public class DerbyChannelMessageStoreQueryProvider extends AbstractChannelMessag
public String getPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE ASC FETCH FIRST ROW ONLY";
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by CREATED_DATE ASC FETCH FIRST ROW ONLY";
"order by CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
}
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
}
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
}
}

28
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/HsqlChannelMessageStoreQueryProvider.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -14,23 +14,45 @@ package org.springframework.integration.jdbc.store.channel;
/**
* @author Gunnar Hillert
* @author Artem Bilan
* @since 2.2
*
*/
public class HsqlChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
@Override
public String getCreateMessageQuery() {
return "INSERT into %PREFIX%CHANNEL_MESSAGE(MESSAGE_ID, GROUP_KEY, REGION, CREATED_DATE, MESSAGE_PRIORITY, MESSAGE_SEQUENCE, MESSAGE_BYTES)"
+ " values (?, ?, ?, ?, ?, NEXT VALUE FOR %PREFIX%MESSAGE_SEQ, ?)";
}
@Override
public String getPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE ASC LIMIT 1";
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by CREATED_DATE ASC LIMIT 1";
"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
}

22
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MySqlChannelMessageStoreQueryProvider.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -14,6 +14,7 @@ package org.springframework.integration.jdbc.store.channel;
/**
* @author Gunnar Hillert
* @author Artem Bilan
* @since 2.2
*/
public class MySqlChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
@@ -22,14 +23,29 @@ public class MySqlChannelMessageStoreQueryProvider extends AbstractChannelMessag
public String getPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE ASC LIMIT 1";
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by CREATED_DATE ASC LIMIT 1";
"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
}

33
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/OracleChannelMessageStoreQueryProvider.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -18,27 +18,48 @@ import org.springframework.jdbc.core.JdbcTemplate;
/**
* Contains Oracle-specific queries for the {@link JdbcChannelMessageStore}.
* Please ensure that the used {@link JdbcTemplate}'s fetchSize property is <code>1</code>.
*
* <p/>
* Fore more details, please see: http://stackoverflow.com/questions/6117254/force-oracle-to-return-top-n-rows-with-skip-locked
*
* @author Gunnar Hillert
* @author Artem Bilan
* @since 2.2
*/
public class OracleChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
@Override
public String getCreateMessageQuery() {
return "INSERT into %PREFIX%CHANNEL_MESSAGE(MESSAGE_ID, GROUP_KEY, REGION, CREATED_DATE, MESSAGE_PRIORITY, MESSAGE_SEQUENCE, MESSAGE_BYTES)"
+ " values (?, ?, ?, ?, ?, %PREFIX%MESSAGE_SEQ.NEXTVAL, ?)";
}
@Override
public String getPollFromGroupExcludeIdsQuery() {
return
"SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE ASC FOR UPDATE SKIP LOCKED";
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE, MESSAGE_SEQUENCE FOR UPDATE SKIP LOCKED";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by CREATED_DATE ASC FOR UPDATE SKIP LOCKED";
"order by CREATED_DATE, MESSAGE_SEQUENCE FOR UPDATE SKIP LOCKED";
}
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FOR UPDATE SKIP LOCKED";
}
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FOR UPDATE SKIP LOCKED";
}
}

22
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/PostgresChannelMessageStoreQueryProvider.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -14,6 +14,7 @@ package org.springframework.integration.jdbc.store.channel;
/**
* @author Gunnar Hillert
* @author Artem Bilan
* @since 2.2
*/
public class PostgresChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
@@ -22,14 +23,29 @@ public class PostgresChannelMessageStoreQueryProvider extends AbstractChannelMes
public String getPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE ASC LIMIT 1 FOR UPDATE";
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by CREATED_DATE ASC LIMIT 1 FOR UPDATE";
"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
}
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
}
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
}
}

5
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-derby.sql
View File

@@ -2,9 +2,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_ID CHAR(36) NOT NULL,
GROUP_KEY CHAR(36) NOT NULL,
CREATED_DATE BIGINT NOT NULL,
MESSAGE_PRIORITY INT,
MESSAGE_SEQUENCE BIGINT NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1),
MESSAGE_BYTES BLOB,
REGION VARCHAR(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
);
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE);
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);

3
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-derby.sql
View File

@@ -1 +1,4 @@
DROP TABLE INT_CHANNEL_MESSAGE;
DROP INDEX INT_CHANNEL_MSG_DATE_IDX;
DROP INDEX INT_CHANNEL_MSG_PRIORITY_IDX;

3
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-hsql.sql
View File

@@ -1 +1,4 @@
DROP TABLE INT_CHANNEL_MESSAGE;
DROP INDEX INT_CHANNEL_MSG_DATE_IDX;
DROP INDEX INT_CHANNEL_MSG_PRIORITY_IDX;
DROP SEQUENCE INT_MESSAGE_SEQ;

2
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-oracle.sql
View File

@@ -1,3 +1,5 @@
DROP INDEX INT_CHANNEL_MSG_DATE_IDX;
DROP TABLE INT_CHANNEL_MESSAGE;
DROP INDEX INT_CHANNEL_MSG_PRIORITY_IDX;
DROP SEQUENCE INT_MESSAGE_SEQ;

4
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-drop-postgres.sql
View File

@@ -1,2 +1,4 @@
DROP TABLE INT_CHANNEL_MESSAGE;
DROP INDEX INT_CHANNEL_MSG_DATE_IDX;
DROP INDEX INT_CHANNEL_MSG_PRIORITY_IDX;
DROP SEQUENCE INT_MESSAGE_SEQ;

7
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-hsql.sql
View File

@@ -2,9 +2,14 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_ID CHAR(36) NOT NULL,
GROUP_KEY CHAR(36) NOT NULL,
CREATED_DATE BIGINT NOT NULL,
MESSAGE_PRIORITY INT,
MESSAGE_SEQUENCE BIGINT NOT NULL,
MESSAGE_BYTES LONGVARBINARY,
REGION VARCHAR(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
);
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE);
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 SEQUENCE INT_MESSAGE_SEQ AS BIGINT START WITH 1 INCREMENT BY 1;

7
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-mysql.sql
View File

@@ -2,10 +2,15 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_ID CHAR(36) NOT NULL,
GROUP_KEY CHAR(36) NOT NULL,
CREATED_DATE BIGINT NOT NULL,
MESSAGE_PRIORITY INT,
MESSAGE_SEQUENCE BIGINT AUTO_INCREMENT UNIQUE,
MESSAGE_BYTES BLOB,
REGION VARCHAR(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
) ENGINE=InnoDB;
ALTER TABLE INT_CHANNEL_MESSAGE
ADD INDEX MSG_INDEX_DATE_IDX USING BTREE (CREATED_DATE ASC) ;
ADD INDEX MSG_INDEX_DATE_IDX USING BTREE (CREATED_DATE, MESSAGE_SEQUENCE);
ALTER TABLE INT_CHANNEL_MESSAGE
ADD INDEX MSG_INDEX_PRIORITY_IDX USING BTREE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);

8
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-oracle.sql
View File

@@ -2,10 +2,14 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_ID CHAR(36) NOT NULL,
GROUP_KEY CHAR(36) NOT NULL,
CREATED_DATE NUMBER(19,0) NOT NULL,
MESSAGE_PRIORITY NUMBER,
MESSAGE_SEQUENCE NUMBER NOT NULL,
MESSAGE_BYTES BLOB,
REGION VARCHAR2(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
);
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX
ON INT_CHANNEL_MESSAGE (CREATED_DATE);
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 SEQUENCE INT_MESSAGE_SEQ START WITH 1 INCREMENT BY 1 NOCACHE NOCYCLE;

9
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/store/channel/schema-postgresql.sql
View File

@@ -2,11 +2,14 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_ID character(36) NOT NULL,
GROUP_KEY character(36) NOT NULL,
CREATED_DATE BIGINT NOT NULL,
MESSAGE_PRIORITY INT,
MESSAGE_SEQUENCE BIGINT NOT NULL DEFAULT nextval('INT_MESSAGE_SEQ'),
MESSAGE_BYTES bytea,
REGION character varying(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
);
CREATE INDEX MSG_INDEX_DATE_IDX
ON INT_CHANNEL_MESSAGE
USING btree (created_date);
CREATE INDEX MSG_INDEX_DATE_IDX ON INT_CHANNEL_MESSAGE USING btree (CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE USING btree (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE SEQUENCE INT_MESSAGE_SEQ START WITH 1 INCREMENT BY 1 NO CYCLE;

20
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractJdbcChannelMessageStoreTests.java
View File

@@ -16,16 +16,20 @@
package org.springframework.integration.jdbc.store.channel;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.*;
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.messaging.Message;
import org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
@@ -36,7 +40,10 @@ import org.springframework.transaction.support.TransactionTemplate;
/**
* @author Gunnar Hillert
*/
public class AbstractJdbcChannelMessageStoreTests {
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext // close at the end after class
public abstract class AbstractJdbcChannelMessageStoreTests {
protected static final String TEST_MESSAGE_GROUP = "AbstractJdbcChannelMessageStoreTests";
@@ -51,6 +58,7 @@ public class AbstractJdbcChannelMessageStoreTests {
@Autowired
protected ChannelMessageStoreQueryProvider queryProvider;
@Before
public void init() throws Exception {
messageStore = new JdbcChannelMessageStore(dataSource);
messageStore.setRegion("AbstractJdbcChannelMessageStoreTests");
@@ -59,11 +67,13 @@ public class AbstractJdbcChannelMessageStoreTests {
messageStore.removeMessageGroup("AbstractJdbcChannelMessageStoreTests");
}
@Test
public void testGetNonExistentMessageFromGroup() throws Exception {
Message<?> result = messageStore.pollMessageFromGroup(TEST_MESSAGE_GROUP);
assertNull(result);
}
@Test
public void testAddAndGet() throws Exception {
final Message<String> message = MessageBuilder.withPayload("Cartman and Kenny")
.setHeader("homeTown", "Southpark")

97
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -13,8 +13,12 @@
package org.springframework.integration.jdbc.store.channel;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.CountDownLatch;
@@ -29,15 +33,24 @@ import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.hamcrest.Matchers;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
@@ -50,7 +63,9 @@ import org.springframework.transaction.support.TransactionTemplate;
* @author Gunnar Hillert
* @author Artem Bilan
*/
abstract class AbstractTxTimeoutMessageStoreTests {
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext // close at the end after class
public abstract class AbstractTxTimeoutMessageStoreTests {
private static final Log log = LogFactory.getLog(AbstractTxTimeoutMessageStoreTests.class);
@@ -79,7 +94,10 @@ abstract class AbstractTxTimeoutMessageStoreTests {
@Autowired
private AtomicInteger errorAtomicInteger;
@Autowired
protected PollableChannel priorityChannel;
@Test
public void test() throws InterruptedException {
int maxMessages = 10;
@@ -106,7 +124,7 @@ abstract class AbstractTxTimeoutMessageStoreTests {
log.info("Done sending " + maxMessages + " messages.");
Assert.assertTrue(String.format("Contdown latch did not count down from " +
Assert.assertTrue(String.format("Countdown latch did not count down from " +
"%s to 0 in %sms.", maxMessages, maxWaitTime), testService.await(maxWaitTime));
Thread.sleep(2000);
@@ -116,6 +134,7 @@ abstract class AbstractTxTimeoutMessageStoreTests {
Assert.assertEquals(Integer.valueOf(0), Integer.valueOf(testService.getDuplicateMessagesCount()));
}
@Test
public void testInt2993IdCacheConcurrency() throws InterruptedException, ExecutionException {
final String groupId = "testInt2993Group";
for (int i = 0; i < 100; i++) {
@@ -175,6 +194,7 @@ abstract class AbstractTxTimeoutMessageStoreTests {
assertTrue(executorService.awaitTermination(5, TimeUnit.SECONDS));
}
@Test
public void testInt3181ConcurrentPolling() throws InterruptedException {
for (int i = 0; i < 10; i++) {
this.first.send(new GenericMessage<Object>("test"));
@@ -185,4 +205,75 @@ abstract class AbstractTxTimeoutMessageStoreTests {
assertEquals(0, errorAtomicInteger.get());
}
@Test
public void testMessageSequenceColumn() throws InterruptedException {
JdbcTemplate jdbcTemplate = new JdbcTemplate(this.dataSource);
String messageGroup = "TEST_MESSAGE_GROUP";
this.jdbcChannelMessageStore.addMessageToGroup(messageGroup, new GenericMessage<Object>("foo"));
// The simple sleep to to be sure that messages are stored with different 'CREATED_DATE'
Thread.sleep(10);
this.jdbcChannelMessageStore.addMessageToGroup(messageGroup, new GenericMessage<Object>("bar"));
List<Map<String, Object>> result =
jdbcTemplate.queryForList("SELECT MESSAGE_SEQUENCE FROM INT_CHANNEL_MESSAGE " +
"WHERE GROUP_KEY = ? ORDER BY CREATED_DATE", UUIDConverter.getUUID(messageGroup).toString());
assertEquals(2, result.size());
Object messageSequence1 = result.get(0).get("MESSAGE_SEQUENCE");
Object messageSequence2 = result.get(1).get("MESSAGE_SEQUENCE");
assertNotNull(messageSequence1);
assertThat(messageSequence1, Matchers.instanceOf(Number.class));
assertNotNull(messageSequence2);
assertThat(messageSequence2, Matchers.instanceOf(Number.class));
assertThat(((Number) messageSequence1).longValue(), Matchers.lessThan(((Number) messageSequence2).longValue()));
this.jdbcChannelMessageStore.removeMessageGroup(messageGroup);
}
@Test
public void testPriorityChannel() throws Exception {
Message<String> message = MessageBuilder.withPayload("1").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, 1).build();
priorityChannel.send(message);
message = MessageBuilder.withPayload("-1").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, -1).build();
priorityChannel.send(message);
message = MessageBuilder.withPayload("3").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, 3).build();
priorityChannel.send(message);
message = MessageBuilder.withPayload("0").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, 0).build();
priorityChannel.send(message);
message = MessageBuilder.withPayload("2").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, 2).build();
priorityChannel.send(message);
message = MessageBuilder.withPayload("none").build();
priorityChannel.send(message);
message = MessageBuilder.withPayload("31").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, 3).build();
priorityChannel.send(message);
Message<?> receive = priorityChannel.receive(1000);
assertNotNull(receive);
assertEquals("3", receive.getPayload());
receive = priorityChannel.receive(1000);
assertNotNull(receive);
assertEquals("31", receive.getPayload());
receive = priorityChannel.receive(1000);
assertNotNull(receive);
assertEquals("2", receive.getPayload());
receive = priorityChannel.receive(1000);
assertNotNull(receive);
assertEquals("1", receive.getPayload());
receive = priorityChannel.receive(1000);
assertNotNull(receive);
assertEquals("0", receive.getPayload());
receive = priorityChannel.receive(1000);
assertNotNull(receive);
assertEquals("-1", receive.getPayload());
receive = priorityChannel.receive(1000);
assertNotNull(receive);
assertEquals("none", receive.getPayload());
}
}

20
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/DerbyTxTimeoutMessageStoreTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -13,34 +13,18 @@
package org.springframework.integration.jdbc.store.channel;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
*
* @author Gunnar Hillert
* @author Artem Bilan
*
*/
@Ignore
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class DerbyTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStoreTests {
@Test
@Override
public void test() throws InterruptedException {
super.test();
}
@Test
@Override
public void testInt3181ConcurrentPolling() throws InterruptedException {
super.testInt3181ConcurrentPolling();
}
}

5
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlJdbcChannelMessageStoreTests-context.xml
View File

@@ -1,11 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-hsql-context.xml"/>
<import resource="classpath:org/springframework/integration/jdbc/store/channel/JdbcChannelMessageStoreTests-context.xml"/>
</beans>

26
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlJdbcChannelMessageStoreTests.java
View File

@@ -16,38 +16,12 @@
package org.springframework.integration.jdbc.store.channel;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gunnar Hillert
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext // close at the end after class
public class HsqlJdbcChannelMessageStoreTests extends AbstractJdbcChannelMessageStoreTests {
@Before
@Override
public void init() throws Exception {
super.init();
}
@Test
@Override
public void testGetNonExistentMessageFromGroup() throws Exception {
super.testGetNonExistentMessageFromGroup();
}
@Test
@Override
public void testAddAndGet() throws Exception {
super.testAddAndGet();
}
}

13
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests-context.xml
View File

@@ -1,19 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:task="http://www.springframework.org/schema/task"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/task/spring-jdbc.xsd
http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-hsql-context.xml"/>
<import resource="classpath:org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml"/>
</beans>

30
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests.java
View File

@@ -12,41 +12,21 @@
*/
package org.springframework.integration.jdbc.store.channel;
import java.util.concurrent.ExecutionException;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
*
* @author Gunnar Hillert
* @author Artem Bilan
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext // close at the end after class
public class HsqlTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStoreTests {
@Test
@Override
public void test() throws InterruptedException {
super.test();
}
@Test
@Override
public void testInt2993IdCacheConcurrency() throws InterruptedException, ExecutionException {
super.testInt2993IdCacheConcurrency();
}
@Test
@Override
public void testInt3181ConcurrentPolling() throws InterruptedException {
super.testInt3181ConcurrentPolling();
}
@Rule
public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
}

8
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/JdbcChannelMessageStoreTests-context.xml
View File

@@ -1,8 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
</beans>

5
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlJdbcChannelMessageStoreTests-context.xml
View File

@@ -1,11 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-mysql-context.xml"/>
<import resource="classpath:org/springframework/integration/jdbc/store/channel/JdbcChannelMessageStoreTests-context.xml"/>
</beans>

25
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlJdbcChannelMessageStoreTests.java
View File

@@ -16,38 +16,15 @@
package org.springframework.integration.jdbc.store.channel;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.ContextConfiguration;
/**
* @author Gunnar Hillert
*/
@Ignore
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class MySqlJdbcChannelMessageStoreTests extends AbstractJdbcChannelMessageStoreTests {
@Before
@Override
public void init() throws Exception {
super.init();
}
@Test
@Override
public void testGetNonExistentMessageFromGroup() throws Exception {
super.testGetNonExistentMessageFromGroup();
}
@Test
@Override
public void testAddAndGet() throws Exception {
super.testAddAndGet();
}
}

13
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlTxTimeoutMessageStoreTests-context.xml
View File

@@ -1,19 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:task="http://www.springframework.org/schema/task"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/task/spring-jdbc.xsd
http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-mysql-context.xml"/>
<import resource="classpath:org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml"/>
</beans>

20
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/MySqlTxTimeoutMessageStoreTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -13,33 +13,17 @@
package org.springframework.integration.jdbc.store.channel;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
*
* @author Gunnar Hillert
* @author Artem Bilan
*
*/
@Ignore
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class MySqlTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStoreTests {
@Test
@Override
public void test() throws InterruptedException {
super.test();
}
@Test
@Override
public void testInt3181ConcurrentPolling() throws InterruptedException {
super.testInt3181ConcurrentPolling();
}
}

13
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/OracleTxTimeoutMessageStoreTests-context.xml
View File

@@ -1,19 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:task="http://www.springframework.org/schema/task"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/task/spring-jdbc.xsd
http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-oracle-context.xml"/>
<import resource="classpath:org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml"/>
</beans>

14
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/OracleTxTimeoutMessageStoreTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -13,26 +13,18 @@
package org.springframework.integration.jdbc.store.channel;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
*
* @author Gunnar Hillert
* @author Artem Bilan
*
*/
@Ignore
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class OracleTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStoreTests {
@Test
@Override
public void test() throws InterruptedException {
super.test();
}
}

13
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/PostgresTxTimeoutMessageStoreTests-context.xml
View File

@@ -1,19 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:task="http://www.springframework.org/schema/task"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/task/spring-jdbc.xsd
http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<import resource="classpath:org/springframework/integration/jdbc/store/channel/DataSource-postgres-context.xml"/>
<import resource="classpath:org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml"/>
</beans>

14
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/PostgresTxTimeoutMessageStoreTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -13,26 +13,18 @@
package org.springframework.integration.jdbc.store.channel;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
*
* @author Gunnar Hillert
* @author Artem Bilan
*
*/
@Ignore
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class PostgresTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStoreTests {
@Test
@Override
public void test() throws InterruptedException {
super.test();
}
}

8
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml
View File

@@ -93,5 +93,13 @@
<int:poller fixed-delay="1000"/>
</int:service-activator>
<bean id="priorityMessageStore" parent="messageStore">
<property name="priorityEnabled" value="true"/>
</bean>
<int:channel id="priorityChannel">
<int:priority-queue message-store="priorityMessageStore"/>
</int:channel>
</beans>

6
spring-integration-jdbc/src/test/resources/log4j.properties
View File

@@ -4,8 +4,8 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m%n
log4j.category.org.springframework.integration=INFO
log4j.category.org.apache.derby=INFO
log4j.category.org.springframework.integration=WARN
log4j.category.org.apache.derby=WARN
log4j.category.org.springframework.jdbc=WARN
log4j.category.org.springframework.integration.jdbc.JdbcChannelMessageStore=INFO
log4j.category.org.springframework.integration.jdbc=WARN

20
src/reference/docbook/channel.xml
View File

@@ -566,15 +566,16 @@ payload to an Integer.
<code>message-store</code> attribute as shown in the next example.
<programlisting language="xml"><![CDATA[<int:channel id="dbBackedChannel">
<int:queue message-store="messageStore"/>
<int:queue message-store="channelStore"/>
</int:channel>
<int-jdbc:message-store id="messageStore" data-source="someDataSource"/>]]></programlisting>
<bean id="channelStore" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
<property name="dataSource" ref="dataSource"/>
<property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
</bean>]]></programlisting>
The above example also shows that <classname>JdbcMessageStore</classname> can be configured with the namespace support
provided by the Spring Integration JDBC module. All you need to do is inject any <classname>javax.sql.DataSource</classname>
instance. The Spring Integration JDBC module also provides schema DDL for most popular databases. These schemas are located in
the <emphasis>org.springframework.integration.jdbc</emphasis> package of that module (spring-integration-jdbc).
The Spring Integration JDBC module also provides schema DDL for a number of popular databases. These schemas are located in
the <emphasis>org.springframework.integration.jdbc.store.channel</emphasis> package of that module (spring-integration-jdbc).
<important>
One important feature is that with any transactional persistent store (e.g., JdbcChannelMessageStore), as long as the poller has a transaction configured,
@@ -662,9 +663,12 @@ payload to an Integer.
<para>
Since <emphasis>version 4.0</emphasis>, the <code>priority-channel</code> child element supports
the <code>message-store</code> option (<code>comparator</code> is not allowed in that case).
The message store must be a <interfacename>ChannelPriorityMessageStore</interfacename> and, in this
The message store must be a <interfacename>PriorityCapableChannelMessageStore</interfacename> and, in this
case, the namespace parser will declare a <classname>QueueChannel</classname> instead of
a <classname>PriorityChannel</classname>. See <xref linkend="channel-configuration-queuechannel"/>.
a <classname>PriorityChannel</classname>. Implementations of the
<classname>PriorityCapableChannelMessageStore</classname> are currently provided for <code>Redis</code>
and <code>JDBC</code>.
See <xref linkend="channel-configuration-queuechannel"/>.
</para>
</section>
<section id="channel-configuration-rendezvouschannel">

67
src/reference/docbook/jdbc.xml
View File

@@ -420,16 +420,12 @@
<section id="jdbc-message-store-channels">
<title>Backing Message Channels</title>
<para>
If you intent backing <emphasis>Message Channels</emphasis> using JDBC,
If you intend backing <emphasis>Message Channels</emphasis> using JDBC,
it is recommended to use the provided <classname>JdbcChannelMessageStore</classname>
implementation instead. It can only be used in conjuntion
implementation instead. It can only be used in conjunction
with <emphasis>Message Channels</emphasis>.
</para>
<note>
The provided <classname>JdbcChannelMessageStore</classname>
implementation is available since <emphasis>Spring Integration 2.2.</emphasis>.
</note>
<para><emphasis>Supported Database</emphasis></para>
<para><emphasis role="bold">Supported Databases</emphasis></para>
<para>
The <classname>JdbcChannelMessageStore</classname> uses database specific
SQL queries to retrieve messages from the database. Therefore, users must
@@ -451,19 +447,27 @@
the <classname>AbstractChannelMessageStoreQueryProvider</classname>
class and provide your own custom queries.
</para>
<para>
Since <emphasis>version 4.0</emphasis>, the <code>MESSAGE_SEQUENCE</code> column has been
added to the table to ensure first-in-first-out (FIFO) queueing even when messages are
stored in the same millisecond.
</para>
<important>
<para>
Generally it is not recommened to use a relational database for the
Generally it is not recommended to use a relational database for the
purpose of queuing. Instead, if possible, consider using either JMS or
AMQP, for which message store implementation are provided as well. For
AMQP backed channels instead. For
further reference please see the following resources:
</para>
<itemizedlist>
<listitem><ulink url="https://www.engineyard.com/blog/2011/5-subtle-ways-youre-using-mysql-as-a-queue-and-why-itll-bite-you/">5 subtle ways you’re using MySQL as a queue, and why it’ll bite you</ulink></listitem>
<listitem><ulink url="http://mikehadlow.blogspot.com/2012/04/database-as-queue-anti-pattern.html">The Database As Queue Anti-Pattern</ulink></listitem>
<listitem><ulink url=
"https://www.engineyard.com/blog/2011/5-subtle-ways-youre-using-mysql-as-a-queue-and-why-itll-bite-you/"
>5 subtle ways you’re using MySQL as a queue, and why it’ll bite you</ulink>.</listitem>
<listitem><ulink url="http://mikehadlow.blogspot.com/2012/04/database-as-queue-anti-pattern.html"
>The Database As Queue Anti-Pattern</ulink>.</listitem>
</itemizedlist>
</important>
<para><emphasis>Concurrent Polling</emphasis></para>
<para><emphasis role="bold">Concurrent Polling</emphasis></para>
<para>
When polling a <emphasis>Message Channel</emphasis>, you have the option
to configure the associated <classname>Poller</classname> with a
@@ -520,7 +524,46 @@
<int:channel id="outputChannel" />
…]]></programlisting>
<para>
<emphasis role="bold">Priority Channel</emphasis>
<para>
Starting with <emphasis>version 4.0</emphasis>, the <classname>JdbcChannelMessageStore</classname>
implements <interfacename>PriorityCapableChannelMessageStore</interfacename> and provides the
<code>priorityEnabled</code> option allowing it to be used as a <code>message-store</code>
reference for <code>priority-queue</code>s. For this purpose, the <code>INT_CHANNEL_MESSAGE</code>
has a <code>MESSAGE_PRIORITY</code> column to store the value of <code>PRIORITY</code> Message header.
In addition, a new <code>MESSAGE_SEQUENCE</code> column is also provided to achieve a robust first-in-first-out
(FIFO) polling mechanism, even when multiple messages are stored with the same priority
in the same millisecond. Messages are polled (selected) from the database with
<code>order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE</code>.
</para>
<note>
<para>
It's not recommended to use the same <classname>JdbcChannelMessageStore</classname> bean
for priority and non-priority queue channel, because <code>priorityEnabled</code> option applies to the
entire store and proper FIFO queue semantics will not be retained for the queue channel.
However the same <code>INT_CHANNEL_MESSAGE</code> table, and even <code>region</code>, can be used for both
<classname>JdbcChannelMessageStore</classname> types. To configure that scenario, simply
extend one message store bean from the other:
</para>
</note>
<programlisting language="xml"><![CDATA[<bean id="channelStore" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
<property name="dataSource" ref="dataSource"/>
<property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
</bean>
<int:channel id="queueChannel">
<int:queue message-store="store"/>
</int:channel>
<bean id="priorityStore" parent="channelStore">
<property name="priorityEnabled" value="true"/>
</bean>
<int:channel id="priorityChannel">
<int:priority-queue message-store="priorityStore"/>
</int:channel>]]></programlisting>
</para>
</section>
<section>
<title>Initializing the Database</title>

9
src/reference/docbook/whats-new.xml
View File

@@ -262,5 +262,14 @@
<xref linkend="sftp-inbound"/>.
</para>
</section>
<section id="4.0-jdbc-cs">
<title>JdbcChannelMessageStore and PriorityChannel</title>
<para>
The <classname>JdbcChannelMessageStore</classname> now implements
<interfacename>PriorityCapableChannelMessageStore</interfacename>, allowing it to be used as
a <code>message-store</code> reference for <code>priority-queue</code>s.
For more information, see <xref linkend="jdbc-message-store-channels"/>.
</para>
</section>
</section>
</chapter>