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

INT-1114: Add MessageGroupStore methods to JdbcMessageStore

Browse Source
This commit is contained in:
David Syer
2010-05-05 17:33:29 +00:00
parent 341a5c02f8
commit 6b1f18723b
7 changed files with 453 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

217
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
View File

@@ -1,17 +1,14 @@
/*
* Copyright 2002-2010 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.
*
* 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;
@@ -21,6 +18,7 @@ import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
@@ -31,27 +29,30 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.integration.core.Message;
import org.springframework.integration.jdbc.util.SerializationUtils;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.AbstractMessageGroupStore;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.SingleColumnRowMapper;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Implementation of {@link MessageStore} using a relational database via JDBC.
* SQL scripts to create the necessary tables are packaged as
* <code>org/springframework/integration/jdbc/schema-*.sql</code>, where
* <code>*</code> is the target database type.
* Implementation of {@link MessageStore} using a relational database via JDBC. SQL scripts to create the necessary
* tables are packaged as <code>org/springframework/integration/jdbc/schema-*.sql</code>, where <code>*</code> is the
* target database type.
*
* @author Dave Syer
* @since 2.0
*/
public class JdbcMessageStore implements MessageStore {
public class JdbcMessageStore extends AbstractMessageGroupStore implements MessageStore {
private static final Log logger = LogFactory.getLog(JdbcMessageStore.class);
@@ -60,29 +61,43 @@ public class JdbcMessageStore implements MessageStore {
*/
public static final String DEFAULT_TABLE_PREFIX = "INT_";
private static final String LIST_MESSAGES_BY_CORRELATION_KEY = "SELECT MESSAGE_ID, CREATED_DATE, CORRELATION_KEY, MESSAGE_BYTES from %PREFIX%MESSAGE where CORRELATION_KEY=?";
private static final String GET_MESSAGE = "SELECT MESSAGE_ID, CREATED_DATE, MESSAGE_BYTES from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?";
private static final String GET_MESSAGE = "SELECT MESSAGE_ID, CREATED_DATE, CORRELATION_KEY, MESSAGE_BYTES from %PREFIX%MESSAGE where MESSAGE_ID=?";
private static final String DELETE_MESSAGE = "DELETE from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?";
private static final String DELETE_MESSAGE = "DELETE from %PREFIX%MESSAGE where MESSAGE_ID=?";
private static final String CREATE_MESSAGE = "INSERT into %PREFIX%MESSAGE(MESSAGE_ID, CREATED_DATE, CORRELATION_KEY, MESSAGE_BYTES)"
private static final String CREATE_MESSAGE = "INSERT into %PREFIX%MESSAGE(MESSAGE_ID, REGION, CREATED_DATE, MESSAGE_BYTES)"
+ " values (?, ?, ?, ?)";
private static final String LIST_UNMARKED_MESSAGES_BY_CORRELATION_KEY = "SELECT MESSAGE_ID, CREATED_DATE, CORRELATION_KEY, MESSAGE_BYTES from %PREFIX%MESSAGE_GROUP where CORRELATION_KEY=? and REGION=? and MARKED=0";
private static final String LIST_MARKED_MESSAGES_BY_CORRELATION_KEY = "SELECT MESSAGE_ID, CREATED_DATE, CORRELATION_KEY, MESSAGE_BYTES from %PREFIX%MESSAGE_GROUP where CORRELATION_KEY=? and REGION=? and MARKED=1";
private static final String GET_MIN_CREATED_DATE_BY_CORRELATION_KEY = "SELECT MIN(CREATED_DATE) from %PREFIX%MESSAGE_GROUP where CORRELATION_KEY=? and REGION=?";
private static final String MARK_MESSAGES_IN_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, MARKED=1 where MARKED=0 and CORRELATION_KEY=? and REGION=?";
private static final String DELETE_MESSAGE_GROUP = "DELETE from %PREFIX%MESSAGE_GROUP where CORRELATION_KEY=? and REGION=?";
private static final String CREATE_MESSAGE_IN_GROUP = "INSERT into %PREFIX%MESSAGE_GROUP(MESSAGE_ID, REGION, CREATED_DATE, CORRELATION_KEY, MARKED, MESSAGE_BYTES)"
+ " values (?, ?, ?, ?, 0, ?)";
private static final String LIST_CORRELATION_KEYS = "SELECT distinct CORRELATION_KEY as CREATED from %PREFIX%MESSAGE_GROUP where REGION=?";
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.
* 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.
*/
public static final String SAVED_KEY = JdbcMessageStore.class.getSimpleName() + ".SAVED";
/**
* The name of the message header that stores a timestamp for the time the
* message was inserted.
* The name of the message header that stores a timestamp for the time the message was inserted.
*/
public static final String CREATED_DATE_KEY = JdbcMessageStore.class.getSimpleName() + ".CREATED_DATE";
private String region = "DEFAULT";
private String tablePrefix = DEFAULT_TABLE_PREFIX;
private JdbcOperations jdbcTemplate;
@@ -91,7 +106,6 @@ public class JdbcMessageStore implements MessageStore {
private MessageMapper mapper = new MessageMapper();
/**
* Convenient constructor for configuration use.
*/
@@ -107,10 +121,8 @@ public class JdbcMessageStore implements MessageStore {
jdbcTemplate = new JdbcTemplate(dataSource);
}
/**
* Replace patterns in the input to produce a valid SQL query. This
* implementation replaces the table prefix.
* Replace patterns in the input to produce a valid SQL query. This implementation replaces the table prefix.
*
* @param base the SQL query to be transformed
* @return a transformed query with replacements
@@ -120,9 +132,8 @@ public class JdbcMessageStore implements MessageStore {
}
/**
* 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}.
* 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
*/
@@ -131,8 +142,17 @@ public class JdbcMessageStore implements MessageStore {
}
/**
* The JDBC {@link DataSource} to use when interacting with the database.
* Either this property can be set or the
* 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 <code>DEFAULT</code>.
*
* @param region the region name to set
*/
public void setRegion(String region) {
this.region = region;
}
/**
* The JDBC {@link DataSource} to use when interacting with the database. Either this property can be set or the
* {@link #setJdbcTemplate(JdbcOperations) jdbcTemplate}.
*
* @param dataSource a {@link DataSource}
@@ -142,8 +162,7 @@ public class JdbcMessageStore implements MessageStore {
}
/**
* The {@link JdbcOperations} to use when interacting with the database.
* Either this property can be set or the
* The {@link JdbcOperations} to use when interacting with the database. Either this property can be set or the
* {@link #setDataSource(DataSource) dataSource}.
*
* @param dataSource a {@link DataSource}
@@ -153,9 +172,8 @@ public class JdbcMessageStore implements MessageStore {
}
/**
* 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.
* 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.
*
* @param lobHandler a {@link LobHandler}
*/
@@ -177,8 +195,8 @@ public class JdbcMessageStore implements MessageStore {
if (message == null) {
return null;
}
int updated = jdbcTemplate.update(getQuery(DELETE_MESSAGE), new Object[] { getKey(id) },
new int[] { Types.VARCHAR });
int updated = jdbcTemplate.update(getQuery(DELETE_MESSAGE), new Object[] { getKey(id), region }, new int[] {
Types.VARCHAR, Types.VARCHAR });
if (updated != 0) {
return message;
}
@@ -186,18 +204,13 @@ public class JdbcMessageStore implements MessageStore {
}
public Message<?> getMessage(UUID id) {
List<Message<?>> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { getKey(id) }, mapper);
List<Message<?>> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { getKey(id), region }, mapper);
if (list.isEmpty()) {
return null;
}
return list.get(0);
}
public List<Message<?>> list(Object correlationId) {
return jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_CORRELATION_KEY),
new Object[] { getKey(correlationId) }, mapper);
}
public <T> Message<T> addMessage(final Message<T> message) {
if (message.getHeaders().containsKey(SAVED_KEY)) {
@SuppressWarnings("unchecked")
@@ -213,38 +226,128 @@ public class JdbcMessageStore implements MessageStore {
Message<T> result = MessageBuilder.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE).setHeader(
CREATED_DATE_KEY, new Long(createdDate)).build();
final String messageId = getKey(result.getHeaders().getId());
final String correlationId = getKey(result.getHeaders().getCorrelationId());
final byte[] messageBytes = SerializationUtils.serialize(result);
jdbcTemplate.update(getQuery(CREATE_MESSAGE), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
logger.debug("Inserting message with id key=" + messageId);
ps.setString(1, messageId);
ps.setTimestamp(2, new Timestamp(createdDate));
ps.setString(3, correlationId);
ps.setString(2, region);
ps.setTimestamp(3, new Timestamp(createdDate));
lobHandler.getLobCreator().setBlobAsBytes(ps, 4, messageBytes);
}
});
return result;
}
public void addMessageToGroup(Object correlationKey, Message<?> message) {
final long createdDate = System.currentTimeMillis();
final String messageId = getKey(message.getHeaders().getId());
final String correlationId = getKey(correlationKey);
final byte[] messageBytes = SerializationUtils.serialize(message);
jdbcTemplate.update(getQuery(CREATE_MESSAGE_IN_GROUP), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
logger.debug("Inserting message with id key=" + messageId + " and created date=" + createdDate);
ps.setString(1, messageId);
ps.setString(2, region);
ps.setTimestamp(3, new Timestamp(createdDate));
ps.setString(4, correlationId);
lobHandler.getLobCreator().setBlobAsBytes(ps, 5, messageBytes);
}
});
}
public MessageGroup getMessageGroup(Object correlationKey) {
String key = getKey(correlationKey);
List<Message<?>> marked = jdbcTemplate.query(getQuery(LIST_MARKED_MESSAGES_BY_CORRELATION_KEY), new Object[] {
key, region }, mapper);
List<Message<?>> unmarked = jdbcTemplate.query(getQuery(LIST_UNMARKED_MESSAGES_BY_CORRELATION_KEY),
new Object[] { key, region }, mapper);
if (marked.isEmpty() && unmarked.isEmpty()) {
return new SimpleMessageGroup(correlationKey);
}
Timestamp date = jdbcTemplate.queryForObject(getQuery(GET_MIN_CREATED_DATE_BY_CORRELATION_KEY),
Timestamp.class, key, region);
Assert.state(date != null, "Could not locate created date for correlationKey=" + correlationKey);
long timestamp = date.getTime();
return new SimpleMessageGroup(unmarked, marked, correlationKey, timestamp);
}
public void markMessageGroup(MessageGroup group) {
final long updatedDate = System.currentTimeMillis();
final String correlationId = getKey(group.getCorrelationKey());
jdbcTemplate.update(getQuery(MARK_MESSAGES_IN_GROUP), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
logger.debug("Marking messages with correlation key=" + correlationId);
ps.setTimestamp(1, new Timestamp(updatedDate));
ps.setString(2, correlationId);
ps.setString(3, region);
}
});
group.mark();
}
public void removeMessageGroup(Object correlationKey) {
final String correlationId = getKey(correlationKey);
jdbcTemplate.update(getQuery(DELETE_MESSAGE_GROUP), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
logger.debug("Marking messages with correlation key=" + correlationId);
ps.setString(1, correlationId);
ps.setString(2, region);
}
});
}
@Override
public Iterator<MessageGroup> iterator() {
@SuppressWarnings("unchecked")
final Iterator<String> iterator = jdbcTemplate.query(getQuery(LIST_CORRELATION_KEYS), new Object[] { region },
new SingleColumnRowMapper(String.class)).iterator();
return new Iterator<MessageGroup>() {
public boolean hasNext() {
return iterator.hasNext();
}
public MessageGroup next() {
return getMessageGroup(iterator.next());
}
public void remove() {
throw new UnsupportedOperationException("Cannot remove MessageGroup from this iterator.");
}
};
}
private String getKey(Object input) {
return input == null ? null : UUIDConverter.getUUID(input).toString();
}
/**
* Convenience class to be used to unpack a message from a result set row.
* Uses column named in the result set to extract the required data, so that
* select clause ordering is unimportant.
* Convenience class to be used to unpack a message from a result set row. Uses column named in the result set to
* extract the required data, so that select clause ordering is unimportant.
*
* @author Dave Syer
*/
private class MessageMapper implements RowMapper<Message<?>> {
public Message<?> mapRow(ResultSet rs, int rowNum) throws SQLException {
Message<?> message = (Message<?>) SerializationUtils.deserialize(
lobHandler.getBlobAsBytes(rs, "MESSAGE_BYTES"));
Message<?> message = (Message<?>) SerializationUtils.deserialize(lobHandler.getBlobAsBytes(rs,
"MESSAGE_BYTES"));
return message;
}
}

15
org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql
View File

@@ -1,6 +1,19 @@
CREATE TABLE INT_MESSAGE (
MESSAGE_ID VARCHAR(100) NOT NULL PRIMARY KEY,
REGION VARCHAR(100),
CREATED_DATE TIMESTAMP NOT NULL,
CORRELATION_KEY VARCHAR(100),
MESSAGE_BYTES BLOB
);
CREATE TABLE INT_MESSAGE_GROUP (
MESSAGE_ID VARCHAR(100) NOT NULL,
CORRELATION_KEY VARCHAR(100) NOT NULL,
REGION VARCHAR(100),
MARKED INTEGER,
CREATED_DATE TIMESTAMP NOT NULL,
UPDATED_DATE TIMESTAMP,
MESSAGE_BYTES BLOB,
constraint MESSAGE_GROUP_PK primary key (MESSAGE_ID, CORRELATION_KEY)
);

58
org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
View File

@@ -5,6 +5,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
import java.util.UUID;
@@ -17,6 +18,7 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.MessageGroup;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.annotation.Transactional;
@@ -55,11 +57,20 @@ public class JdbcMessageStoreTests {
assertNotNull(result.getHeaders().get(JdbcMessageStore.CREATED_DATE_KEY));
}
@Test
@Transactional
public void testAddAndGetWithDifferentRegion() throws Exception {
Message<String> message = MessageBuilder.withPayload("foo").build();
Message<String> saved = messageStore.addMessage(message);
messageStore.setRegion("FOO");
Message<?> result = messageStore.getMessage(saved.getHeaders().getId());
assertNull(result);
}
@Test
@Transactional
public void testAddAndUpdate() throws Exception {
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(
"X").build();
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId("X").build();
message = messageStore.addMessage(message);
message = MessageBuilder.fromMessage(message).setCorrelationId("Y").build();
message = messageStore.addMessage(message);
@@ -87,15 +98,6 @@ public class JdbcMessageStoreTests {
assertNotNull(messageStore.getMessage(saved.getHeaders().getId()));
}
@Test
@Transactional
public void testAddAndListByCorrelationId() throws Exception {
String correlationId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build();
messageStore.addMessage(message);
assertEquals(1, messageStore.list(correlationId).size());
}
@Test
@Transactional
public void testAddAndDelete() throws Exception {
@@ -104,4 +106,38 @@ public class JdbcMessageStoreTests {
assertNotNull(messageStore.removeMessage(message.getHeaders().getId()));
}
@Test
@Transactional
public void testAddAndGetMessageGroup() throws Exception {
String correlationId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build();
long now = System.currentTimeMillis();
messageStore.addMessageToGroup(correlationId, message);
MessageGroup group = messageStore.getMessageGroup(correlationId);
assertEquals(1, group.size());
assertTrue("Timestamp too early: " + group.getTimestamp() + "<" + now, group.getTimestamp() >= now);
}
@Test
@Transactional
public void testAddAndMarkMessageGroup() throws Exception {
String correlationId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build();
messageStore.addMessageToGroup(correlationId, message);
MessageGroup group = messageStore.getMessageGroup(correlationId);
messageStore.markMessageGroup(group);
assertEquals(1, group.getMarked().size());
}
@Test
@Transactional
public void testExpireMessageGroup() throws Exception {
String correlationId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build();
messageStore.addMessageToGroup(correlationId, message);
messageStore.expireMessageGroups(-10000);
MessageGroup group = messageStore.getMessageGroup(correlationId);
assertEquals(0, group.getMarked().size());
}
}

94
org.springframework.integration/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java Normal file
View File

@@ -0,0 +1,94 @@
/*
* Copyright 2002-2010 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.store;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedHashSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* @author Dave Syer
*
* @since 2.0
*
*/
public abstract class AbstractMessageGroupStore implements MessageGroupStore, Iterable<MessageGroup> {
protected final Log logger = LogFactory.getLog(getClass());
private Collection<MessageGroupCallback> expiryCallbacks = new LinkedHashSet<MessageGroupCallback>();
/**
*
*/
public AbstractMessageGroupStore() {
super();
}
/**
* Convenient injection point for expiry callbacks in the message store. Each of the callbacks provided will simply
* be registered with the store using {@link #registerExpiryCallback(MessageGroupCallback)}.
*
* @param expiryCallbacks the expiry callbacks to add
*/
public void setExpiryCallbacks(Collection<MessageGroupCallback> expiryCallbacks) {
for (MessageGroupCallback callback : expiryCallbacks) {
registerExpiryCallback(callback);
}
}
public void registerExpiryCallback(MessageGroupCallback callback) {
expiryCallbacks.add(callback);
}
public int expireMessageGroups(long timeout) {
int count = 0;
long threshold = System.currentTimeMillis() - timeout;
for (MessageGroup group : this) {
if (group.getTimestamp() < threshold) {
count++;
expire(group);
removeMessageGroup(group.getCorrelationKey());
}
}
return count;
}
public abstract Iterator<MessageGroup> iterator();
private void expire(MessageGroup group) {
RuntimeException exception = null;
for (MessageGroupCallback callback : expiryCallbacks) {
try {
callback.execute(group);
} catch (RuntimeException e) {
if (exception == null) {
exception = e;
}
logger.error("Exception in expiry callback", e);
}
}
if (exception != null) {
throw exception;
}
}
}

64
org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java
View File

@@ -14,15 +14,15 @@
package org.springframework.integration.store;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import org.springframework.integration.core.Message;
/**
* Represents a mutable group of correlated messages that is bound to a certain
* {@link MessageStore} and correlation key. The group will grow during its
* lifetime, when messages are <code>add</code>ed to it. <strong>This is not
* thread safe and should not be used for long running aggregations</strong>.
* Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and correlation
* key. The group will grow during its lifetime, when messages are <code>add</code>ed to it. <strong>This is not thread
* safe and should not be used for long running aggregations</strong>.
*
* @author Iwein Fuld
* @author Oleg Zhurakousky
@@ -41,18 +41,24 @@ public class SimpleMessageGroup implements MessageGroup {
private final long timestamp;
public SimpleMessageGroup(Object correlationKey) {
this.correlationKey = correlationKey;
this.timestamp = System.currentTimeMillis();
this(Collections.<Message<?>>emptyList(), Collections.<Message<?>>emptyList(), correlationKey, System.currentTimeMillis());
}
public SimpleMessageGroup(Collection<? extends Message<?>> originalMessages,
Object correlationKey) {
this(correlationKey);
for (Message<?> message : originalMessages) {
add(message);
public SimpleMessageGroup(Collection<? extends Message<?>> unmarked, Object correlationKey) {
this(unmarked, Collections.<Message<?>>emptyList(), correlationKey, System.currentTimeMillis());
}
public SimpleMessageGroup(Collection<? extends Message<?>> unmarked, Collection<? extends Message<?>> marked, Object correlationKey, long timestamp) {
this.correlationKey = correlationKey;
this.timestamp = timestamp;
for (Message<?> message : unmarked) {
addUnmarked(message);
}
for (Message<?> message : marked) {
addMarked(message);
}
}
public SimpleMessageGroup(MessageGroup template) {
this.correlationKey = template.getCorrelationKey();
this.marked.addAll(template.getMarked());
@@ -65,6 +71,10 @@ public class SimpleMessageGroup implements MessageGroup {
}
public boolean add(Message<?> message) {
return addUnmarked(message);
}
private boolean addUnmarked(Message<?> message) {
if (isMember(message)) {
return false;
}
@@ -72,6 +82,14 @@ public class SimpleMessageGroup implements MessageGroup {
return true;
}
private boolean addMarked(Message<?> message) {
if (isMember(message)) {
return false;
}
this.marked.add(message);
return true;
}
public Collection<Message<?>> getUnmarked() {
return unmarked;
}
@@ -110,25 +128,21 @@ public class SimpleMessageGroup implements MessageGroup {
}
public Message<?> getOne() {
return unmarked.isEmpty() ? (marked.isEmpty() ? null : marked
.iterator().next()) : unmarked.iterator().next();
return unmarked.isEmpty() ? (marked.isEmpty() ? null : marked.iterator().next()) : unmarked.iterator().next();
}
/**
* This method determines whether messages have been added to this group
* that supersede the given message based on its sequence id. This can be
* helpful to avoid ending up with sequences larger than their required
* sequence size or sequences that are missing certain sequence numbers.
* This method determines whether messages have been added to this group that supersede the given message based on
* its sequence id. This can be helpful to avoid ending up with sequences larger than their required sequence size
* or sequences that are missing certain sequence numbers.
*/
private boolean isMember(Message<?> message) {
if (size() == 0) {
return false;
}
Integer messageSequenceNumber = message.getHeaders()
.getSequenceNumber();
Integer messageSequenceNumber = message.getHeaders().getSequenceNumber();
if (messageSequenceNumber != null && messageSequenceNumber > 0) {
Integer messageSequenceSize = message.getHeaders()
.getSequenceSize();
Integer messageSequenceSize = message.getHeaders().getSequenceSize();
if (!messageSequenceSize.equals(getSequenceSize())
|| containsSequenceNumber(unmarked, messageSequenceNumber)
|| containsSequenceNumber(marked, messageSequenceNumber)) {
@@ -138,11 +152,9 @@ public class SimpleMessageGroup implements MessageGroup {
return false;
}
private boolean containsSequenceNumber(Collection<Message<?>> messages,
Integer messageSequenceNumber) {
private boolean containsSequenceNumber(Collection<Message<?>> messages, Integer messageSequenceNumber) {
for (Message<?> member : messages) {
Integer memberSequenceNumber = member.getHeaders()
.getSequenceNumber();
Integer memberSequenceNumber = member.getHeaders().getSequenceNumber();
if (messageSequenceNumber.equals(memberSequenceNumber)) {
return true;
}

65
org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -13,14 +13,12 @@
package org.springframework.integration.store;
import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.HashSet;
import java.util.Iterator;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.util.UpperBound;
@@ -36,20 +34,16 @@ import org.springframework.util.Assert;
*
* @since 2.0
*/
public class SimpleMessageStore implements MessageStore, MessageGroupStore {
private static final Log logger = LogFactory.getLog(SimpleMessageStore.class);
public class SimpleMessageStore extends AbstractMessageGroupStore implements MessageStore, MessageGroupStore {
private final ConcurrentMap<UUID, Message<?>> idToMessage;
private final ConcurrentMap<Object, SimpleMessageGroup> correlationToMessageGroup;
final ConcurrentMap<Object, SimpleMessageGroup> correlationToMessageGroup;
private final UpperBound individualUpperBound;
private final UpperBound groupUpperBound;
private Collection<MessageGroupCallback> expiryCallbacks = new LinkedHashSet<MessageGroupCallback>();
/**
* Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given
* capacity is less than 1. The capacities are applied independently to messages stored via
@@ -78,18 +72,6 @@ public class SimpleMessageStore implements MessageStore, MessageGroupStore {
this(0);
}
/**
* Convenient injection point for expiry callbacks in the message store. Each of the callbacks provided will simply
* be registered with the store using {@link #registerExpiryCallback(MessageGroupCallback)}.
*
* @param expiryCallbacks the expiry callbacks to add
*/
public void setExpiryCallbacks(Collection<MessageGroupCallback> expiryCallbacks) {
for (MessageGroupCallback callback : expiryCallbacks) {
registerExpiryCallback(callback);
}
}
public <T> Message<T> addMessage(Message<T> message) {
if (!individualUpperBound.tryAcquire(0)) {
throw new MessagingException(this.getClass().getSimpleName()
@@ -140,42 +122,9 @@ public class SimpleMessageStore implements MessageStore, MessageGroupStore {
correlationToMessageGroup.remove(correlationId);
}
public void registerExpiryCallback(MessageGroupCallback callback) {
expiryCallbacks.add(callback);
}
public int expireMessageGroups(long timeout) {
int count = 0;
long threshold = System.currentTimeMillis() - timeout;
for (MessageGroup group : correlationToMessageGroup.values()) {
if (group.getTimestamp() < threshold) {
count++;
expire(group);
removeMessageGroup(group.getCorrelationKey());
}
}
return count;
}
private void expire(MessageGroup group) {
RuntimeException exception = null;
for (MessageGroupCallback callback : expiryCallbacks) {
try {
callback.execute(group);
} catch (RuntimeException e) {
if (exception == null) {
exception = e;
}
logger.error("Exception in expiry callback", e);
}
}
if (exception != null) {
throw exception;
}
@Override
public Iterator<MessageGroup> iterator() {
return new HashSet<MessageGroup>(correlationToMessageGroup.values()).iterator();
}
private SimpleMessageGroup getMessageGroupInternal(Object correlationId) {

93
org.springframework.integration/src/test/java/org/springframework/integration/store/MessageStoreTests.java Normal file
View File

@@ -0,0 +1,93 @@
/*
* Copyright 2002-2010 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.store;
import static org.junit.Assert.assertEquals;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import org.junit.Test;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.test.util.ReflectionTestUtils;
/**
* @author Dave Syer
*/
public class MessageStoreTests {
@Test
public void shouldRegisterCallbacks() throws Exception {
TestMessageStore store = new TestMessageStore();
store.setExpiryCallbacks(Arrays.<MessageGroupCallback>asList(new MessageGroupCallback() {
public void execute(MessageGroup group) {
}
}));
assertEquals(1, ((Collection<?>)ReflectionTestUtils.getField(store, "expiryCallbacks")).size());
}
@Test
public void shouldExpireMessageGroup() throws Exception {
TestMessageStore store = new TestMessageStore();
final List<String> list = new ArrayList<String>();
store.registerExpiryCallback(new MessageGroupCallback() {
public void execute(MessageGroup group) {
list.add(group.getOne().getPayload().toString());
}
});
store.expireMessageGroups(-10000);
assertEquals("[foo]", list.toString());
assertEquals(0, store.getMessageGroup("bar").size());
}
private static class TestMessageStore extends AbstractMessageGroupStore {
MessageGroup testMessages = new SimpleMessageGroup(Arrays.asList(new StringMessage("foo")), "bar");
private boolean removed = false;
@Override
public Iterator<MessageGroup> iterator() {
return Arrays.asList(testMessages).iterator();
}
public void addMessageToGroup(Object correlationKey, Message<?> message) {
}
public MessageGroup getMessageGroup(Object correlationKey) {
return removed ? new SimpleMessageGroup(correlationKey) : testMessages;
}
public void markMessageGroup(MessageGroup group) {
}
public void removeMessageGroup(Object correlationKey) {
if (correlationKey.equals(testMessages.getCorrelationKey())) {
removed = true;
}
}
}
}