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

RESOLVED - issue INT-1010, INT-1050, INT-1100

Browse Source
This commit is contained in:
David Syer
2010-04-26 16:30:22 +00:00
parent dd4afd8cb5
commit 193d9b6002
4 changed files with 38 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -49,8 +49,6 @@ public class JdbcMessageStore implements MessageStore {
private static final String LIST_MESSAGES_BY_CORRELATION_KEY = "SELECT STORE_ID, MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES, VERSION from %PREFIX%MESSAGE where CORRELATION_KEY=?";
private static final String LIST_ALL_MESSAGES = "SELECT STORE_ID, MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES, VERSION from %PREFIX%MESSAGE";
private static final String GET_MESSAGE = "SELECT STORE_ID, MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES, VERSION from %PREFIX%MESSAGE where MESSAGE_ID=?";
private static final String DELETE_MESSAGE = "DELETE from %PREFIX%MESSAGE where MESSAGE_ID=?";
@@ -205,10 +203,6 @@ public class JdbcMessageStore implements MessageStore {
return list.get(0);
}
public List<Message<?>> list() {
return jdbcTemplate.query(getQuery(LIST_ALL_MESSAGES), new MessageMapper());
}
public List<Message<?>> list(Object correlationId) {
return jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_CORRELATION_KEY), new Object[] { getKey(correlationId) },
new MessageMapper());

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

@@ -15,11 +15,9 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.jdbc.SimpleJdbcTestUtils;
import org.springframework.transaction.annotation.Transactional;
@ContextConfiguration
@@ -36,12 +34,9 @@ public class JdbcMessageStoreTests {
private JdbcMessageStore messageStore;
private SimpleJdbcTemplate jdbcTemplate;
@Before
public void init() {
messageStore = new JdbcMessageStore(dataSource, messageIncrementer);
jdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
@Test
@@ -71,15 +66,6 @@ public class JdbcMessageStoreTests {
assertEquals("Y", messageStore.get(message.getHeaders().getId()).getHeaders().getCorrelationId());
}
@Test
@Transactional
public void testAddAndList() throws Exception {
Message<String> message = MessageBuilder.withPayload("foo").build();
messageStore.put(message);
assertEquals(1, SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "INT_MESSAGE"));
assertEquals(1, messageStore.list().size());
}
@Test
@Transactional
public void testAddAndListByCorrelationId() throws Exception {

5
org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java
View File

@@ -55,11 +55,6 @@ public interface MessageStore {
*/
Message<?> delete(UUID id);
/**
* Return all Messages currently in the MessageStore.
*/
List<Message<?>> list();
/**
* Return all Messages currently in the MessageStore that
* contain the provided correlationId header value.

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

@@ -29,25 +29,27 @@ import org.springframework.integration.util.UpperBound;
import org.springframework.util.Assert;
/**
* Map-based implementation of {@link MessageStore} that enforces a maximum capacity.
*
* Map-based implementation of {@link MessageStore} that enforces a maximum
* capacity.
*
* @author Iwein Fuld
* @author Mark Fisher
* @since 2.0
*/
public class SimpleMessageStore implements MessageStore {
private final Map<UUID, Message<?>> map;
private final Map<UUID, Message<?>> map;
private final UpperBound upperBound;
/**
* Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited
* size if the given capacity is less than 1.
* Creates a SimpleMessageStore with a maximum size limited by the given
* capacity, or unlimited size if the given capacity is less than 1.
*/
public SimpleMessageStore(int capacity) {
this.map = new ConcurrentHashMap<UUID, Message<?>>();
this.upperBound = new UpperBound(capacity);
}
this.map = new ConcurrentHashMap<UUID, Message<?>>();
this.upperBound = new UpperBound(capacity);
}
/**
* Creates a SimpleMessageStore with unlimited capacity
@@ -56,48 +58,43 @@ public class SimpleMessageStore implements MessageStore {
this(0);
}
@SuppressWarnings("unchecked")
public <T> Message<T> put(Message<T> message) {
if (!upperBound.tryAcquire(0)){
throw new MessagingException(this.getClass().getSimpleName() +
" was out of capacity at, try constructing it with a larger capacity.");
public <T> Message<T> put(Message<T> message) {
if (!upperBound.tryAcquire(0)) {
throw new MessagingException(this.getClass().getSimpleName()
+ " was out of capacity at, try constructing it with a larger capacity.");
}
return (Message<T>) this.map.put(message.getHeaders().getId(), message);
}
return (Message<T>) this.map.put(message.getHeaders().getId(), message);
}
public Message<?> get(UUID key) {
return (key != null) ? this.map.get(key) : null;
}
public Message<?> get(UUID key) {
return (key != null) ? this.map.get(key) : null;
}
public List<Message<?>> list() {
return new ArrayList<Message<?>>(this.map.values());
}
public Message<?> delete(UUID key) {
public Message<?> delete(UUID key) {
if (key != null) {
upperBound.release();
return this.map.remove(key);
}
else return null;
}
else
return null;
}
public int size() {
return this.map.size();
}
public int size() {
return this.map.size();
}
public List<Message<?>> list(Object correlationKey) {
Assert.notNull(correlationKey, "'correlationKey' must not be null");
List<Message<?>> matched = new ArrayList<Message<?>>();
Collection<Message<?>> values = map.values();
for (Message<?> message : values) {
Object correlationId = message.getHeaders().getCorrelationId();
if (correlationId != null && correlationId.equals(correlationKey)) {
matched.add(message);
}
}
return matched;
}
public List<Message<?>> list(Object correlationKey) {
Assert.notNull(correlationKey, "'correlationKey' must not be null");
List<Message<?>> matched = new ArrayList<Message<?>>();
Collection<Message<?>> values = map.values();
for (Message<?> message : values) {
Object correlationId = message.getHeaders().getCorrelationId();
if (correlationId != null && correlationId.equals(correlationKey)) {
matched.add(message);
}
}
return matched;
}
}