diff --git a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index 32c47f09b5..358d95a970 100644 --- a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -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 - * org/springframework/integration/jdbc/schema-*.sql, where - * * 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 org/springframework/integration/jdbc/schema-*.sql, where * 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 DEFAULT. + * + * @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> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { getKey(id) }, mapper); + List> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { getKey(id), region }, mapper); if (list.isEmpty()) { return null; } return list.get(0); } - public List> list(Object correlationId) { - return jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_CORRELATION_KEY), - new Object[] { getKey(correlationId) }, mapper); - } - public Message addMessage(final Message message) { if (message.getHeaders().containsKey(SAVED_KEY)) { @SuppressWarnings("unchecked") @@ -213,38 +226,128 @@ public class JdbcMessageStore implements MessageStore { Message 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> marked = jdbcTemplate.query(getQuery(LIST_MARKED_MESSAGES_BY_CORRELATION_KEY), new Object[] { + key, region }, mapper); + List> 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 iterator() { + + @SuppressWarnings("unchecked") + final Iterator iterator = jdbcTemplate.query(getQuery(LIST_CORRELATION_KEYS), new Object[] { region }, + new SingleColumnRowMapper(String.class)).iterator(); + + return new Iterator() { + + 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> { 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; } } diff --git a/org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql b/org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql index 71bdeb7b5f..e8bf2a6cdb 100644 --- a/org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql +++ b/org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql @@ -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) +); + + diff --git a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index b5c5cdc358..c5a0a7b984 100644 --- a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -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 message = MessageBuilder.withPayload("foo").build(); + Message saved = messageStore.addMessage(message); + messageStore.setRegion("FOO"); + Message result = messageStore.getMessage(saved.getHeaders().getId()); + assertNull(result); + } + @Test @Transactional public void testAddAndUpdate() throws Exception { - Message message = MessageBuilder.withPayload("foo").setCorrelationId( - "X").build(); + Message 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 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 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 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 message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build(); + messageStore.addMessageToGroup(correlationId, message); + messageStore.expireMessageGroups(-10000); + MessageGroup group = messageStore.getMessageGroup(correlationId); + assertEquals(0, group.getMarked().size()); + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java new file mode 100644 index 0000000000..e53ff3d6c9 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java @@ -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 { + + protected final Log logger = LogFactory.getLog(getClass()); + + private Collection expiryCallbacks = new LinkedHashSet(); + + /** + * + */ + 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 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 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; + } + + } + +} \ No newline at end of file diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java index 8b56b1c0e7..3dc466a1d2 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java @@ -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 added to it. This is not - * thread safe and should not be used for long running aggregations. + * 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 added to it. This is not thread + * safe and should not be used for long running aggregations. * * @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.>emptyList(), Collections.>emptyList(), correlationKey, System.currentTimeMillis()); } - public SimpleMessageGroup(Collection> originalMessages, - Object correlationKey) { - this(correlationKey); - for (Message message : originalMessages) { - add(message); + public SimpleMessageGroup(Collection> unmarked, Object correlationKey) { + this(unmarked, Collections.>emptyList(), correlationKey, System.currentTimeMillis()); + } + + public SimpleMessageGroup(Collection> unmarked, Collection> 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> 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> messages, - Integer messageSequenceNumber) { + private boolean containsSequenceNumber(Collection> messages, Integer messageSequenceNumber) { for (Message member : messages) { - Integer memberSequenceNumber = member.getHeaders() - .getSequenceNumber(); + Integer memberSequenceNumber = member.getHeaders().getSequenceNumber(); if (messageSequenceNumber.equals(memberSequenceNumber)) { return true; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index 219d6ab9e5..73837dd87a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -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> idToMessage; - private final ConcurrentMap correlationToMessageGroup; + final ConcurrentMap correlationToMessageGroup; private final UpperBound individualUpperBound; private final UpperBound groupUpperBound; - private Collection expiryCallbacks = new LinkedHashSet(); - /** * 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 expiryCallbacks) { - for (MessageGroupCallback callback : expiryCallbacks) { - registerExpiryCallback(callback); - } - } - public Message addMessage(Message 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 iterator() { + return new HashSet(correlationToMessageGroup.values()).iterator(); } private SimpleMessageGroup getMessageGroupInternal(Object correlationId) { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/store/MessageStoreTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/store/MessageStoreTests.java new file mode 100644 index 0000000000..7e9f54394b --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/store/MessageStoreTests.java @@ -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.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 list = new ArrayList(); + 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 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; + } + } + + } + +}