diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java index 885523d6ce..402e4d2c10 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java @@ -12,6 +12,10 @@ */ package org.springframework.integration.aggregator; +import java.util.Iterator; +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; @@ -21,10 +25,6 @@ import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; -import java.util.Iterator; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; - /** * This Endpoint serves as a barrier for messages that should not be processed yet. The decision when a message can be * processed is delegated to a {@link org.springframework.integration.aggregator.ReleaseStrategy ReleaseStrategy}. @@ -37,7 +37,7 @@ import java.util.concurrent.ConcurrentMap; * * @author Iwein Fuld */ -public class CorrelatingMessageBarrier extends AbstractMessageHandler implements MessageSource { +public class CorrelatingMessageBarrier extends AbstractMessageHandler implements MessageSource { private static final Log log = LogFactory.getLog(CorrelatingMessageBarrier.class); private CorrelationStrategy correlationStrategy; @@ -86,7 +86,7 @@ public class CorrelatingMessageBarrier extends AbstractMessageHandler implements } - public Message receive() { + public Message receive() { for (Object key : correlationLocks.keySet()) { Object lock = getLock(key); synchronized (lock) { @@ -106,7 +106,9 @@ public class CorrelatingMessageBarrier extends AbstractMessageHandler implements } else { remove(key); } - return nextMessage; + @SuppressWarnings("unchecked") + Message result = (Message) nextMessage; + return result; } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 3f4b752b89..fb23937b8c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -108,7 +108,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements public void setMessageStore(MessageGroupStore store) { this.messageStore = store; store.registerMessageGroupExpiryCallback(new MessageGroupCallback() { - public void execute(MessageGroup group) { + public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { forceComplete(group); } }); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java index 531a9ad073..c307cb8ff8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java @@ -62,7 +62,6 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It if (group.getTimestamp() < threshold) { count++; expire(group); - removeMessageGroup(group.getCorrelationKey()); } } return count; @@ -76,7 +75,7 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It for (MessageGroupCallback callback : expiryCallbacks) { try { - callback.execute(group); + callback.execute(this, group); } catch (RuntimeException e) { if (exception == null) { exception = e; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupCallback.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupCallback.java index 320b0940a3..02cfc158d2 100755 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupCallback.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupCallback.java @@ -21,6 +21,6 @@ package org.springframework.integration.store; */ public interface MessageGroupCallback { - void execute(MessageGroup group); + void execute(MessageGroupStore messageGroupStore, MessageGroup group); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java new file mode 100644 index 0000000000..4f78903f0d --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java @@ -0,0 +1,192 @@ +/* + * Copyright 2002-2008 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.AbstractQueue; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Iterator; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.springframework.integration.core.Message; + +/** + * A {@link BlockingQueue} that is backed by a {@link MessageGroupStore}. Can be used to ensure guaranteed delivery in + * the face of transaction rollback (assuming the store is transactional) and also to ensure messages are not lost if + * the process dies (assuming the store is durable). To use the queue across process re-starts, the same correlation key + * must be provided, so it needs to be unique but identifiable with a single logical instance of the queue. + * + * @author Dave Syer + * @since 2.0 + * + */ +public class MessageGroupQueue extends AbstractQueue> implements BlockingQueue> { + + private static final int DEFAULT_CAPACITY = Integer.MAX_VALUE; + + private final MessageGroupStore messageGroupStore; + + private final Object correlationKey; + + private final int capacity; + + // This one could be a global semaphore + private Object storeLock = new Object(); + + // This one only needs to be local + private Object writeLock = new Object(); + + // This one only needs to be local + private Object readLock = new Object(); + + public MessageGroupQueue(MessageGroupStore messageGroupStore, Object correlationKey) { + this(messageGroupStore, correlationKey, DEFAULT_CAPACITY); + } + + public MessageGroupQueue(MessageGroupStore messageGroupStore, Object correlationKey, int capacity) { + this.messageGroupStore = messageGroupStore; + this.correlationKey = correlationKey; + this.capacity = capacity; + } + + public Iterator> iterator() { + return getUnmarked().iterator(); + } + + public int size() { + return getUnmarked().size(); + } + + public boolean offer(Message e) { + if (messageGroupStore.getMessageGroup(correlationKey).size() >= capacity) { + return false; + } + synchronized (storeLock) { + messageGroupStore.addMessageToGroup(correlationKey, e); + } + synchronized (readLock) { + readLock.notifyAll(); + } + return true; + } + + public Message peek() { + Collection> unmarked = getUnmarked(); + if (unmarked.isEmpty()) { + return null; + } + return unmarked.iterator().next(); + } + + public Message poll() { + Message result; + synchronized (storeLock) { + Collection> unmarked = getUnmarked(); + if (unmarked.isEmpty()) { + return null; + } + result = unmarked.iterator().next(); + messageGroupStore.removeMessageFromGroup(correlationKey, result); + } + synchronized (writeLock) { + writeLock.notifyAll(); + } + return result; + } + + public int drainTo(Collection> c) { + Collection> unmarked; + synchronized (storeLock) { + unmarked = getUnmarked(); + c.addAll(unmarked); + messageGroupStore.markMessageGroup(messageGroupStore.getMessageGroup(correlationKey)); + } + synchronized (writeLock) { + writeLock.notifyAll(); + } + return unmarked.size(); + } + + public int drainTo(Collection> c, int maxElements) { + ArrayList> list = new ArrayList>(); + synchronized (storeLock) { + Iterator> unmarked = getUnmarked().iterator(); + for (int i = 0; i < maxElements && unmarked.hasNext(); i++) { + Message message = unmarked.next(); + messageGroupStore.removeMessageFromGroup(correlationKey, message); + list.add(message); + } + } + synchronized (writeLock) { + writeLock.notifyAll(); + } + c.addAll(list); + return list.size(); + } + + public boolean offer(Message e, long timeout, TimeUnit unit) throws InterruptedException { + if (!offer(e)) { + synchronized (writeLock) { + writeLock.wait(TimeUnit.MILLISECONDS.convert(timeout, unit)); + } + } + return offer(e); + } + + public Message poll(long timeout, TimeUnit unit) throws InterruptedException { + Message message = poll(); + if (message != null) { + return message; + } + long threshold = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(timeout, unit); + while (message == null && System.currentTimeMillis() < threshold) { + synchronized (readLock) { + readLock.wait(threshold - System.currentTimeMillis()); + } + message = poll(); + } + return message; + } + + public void put(Message e) throws InterruptedException { + while (!offer(e)) { + synchronized (writeLock) { + writeLock.wait(); + } + } + } + + public int remainingCapacity() { + return capacity - messageGroupStore.getMessageGroup(correlationKey).size(); + } + + public Message take() throws InterruptedException { + Message message = poll(); + while (message == null) { + synchronized (readLock) { + readLock.wait(); + } + message = poll(); + } + return message; + } + + private Collection> getUnmarked() { + return messageGroupStore.getMessageGroup(correlationKey).getUnmarked(); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java index 2e42386f3c..e4140e7c80 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java @@ -54,15 +54,15 @@ public class CorrelatingMessageBarrierTest { @Test public void shouldPassMessage() { - Message message = testMessage(); + Message message = testMessage(); barrier.handleMessage(message); assertThat(barrier.receive(), is(message)); } @Test public void shouldRemoveKeyWithoutLockingOnEmptyQueue() throws InterruptedException { - Message message = testMessage(); - Message message2 = testMessage(); + Message message = testMessage(); + Message message2 = testMessage(); barrier.handleMessage(message); verify(correlationStrategy).getCorrelationKey(message); assertThat(barrier.receive(), is(notNullValue())); @@ -95,7 +95,7 @@ public class CorrelatingMessageBarrierTest { } } - private void sendAsynchronously(final MessageHandler handler, final Message message, final CountDownLatch start, final CountDownLatch sent) { + private void sendAsynchronously(final MessageHandler handler, final Message message, final CountDownLatch start, final CountDownLatch sent) { Executors.newSingleThreadExecutor().execute(new Runnable() { public void run() { try { @@ -110,8 +110,8 @@ public class CorrelatingMessageBarrierTest { } - private Message testMessage() { - return MessageBuilder.withPayload("payload").build(); + private Message testMessage() { + return MessageBuilder.withPayload((Object)"payload").build(); } @@ -145,6 +145,7 @@ public class CorrelatingMessageBarrierTest { } } + @SuppressWarnings("unused") public void releaseAll() { for (Semaphore semaphore : keyLocks.values()) { semaphore.release(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java index a74290c9a2..9978e13dde 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java @@ -22,7 +22,6 @@ import static org.junit.Assert.assertTrue; import org.junit.Test; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.SimpleMessageGroup; /** diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategyTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategyTests.java index f8d5fc8bc1..f3365b4e1e 100755 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategyTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategyTests.java @@ -22,7 +22,6 @@ import static org.junit.Assert.assertTrue; import org.junit.Test; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.SimpleMessageGroup; /** diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java new file mode 100644 index 0000000000..b2dd679207 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java @@ -0,0 +1,154 @@ +/* + * Copyright 2002-2008 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 static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Test; +import org.springframework.integration.core.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Dave Syer + * @since 2.0 + * + */ +public class MessageGroupQueueTests { + + static final Log logger = LogFactory.getLog(MessageGroupQueueTests.class); + + @Test + public void testPutAndPoll() throws Exception { + MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), "FOO"); + queue.put(new StringMessage("foo")); + Message result = queue.poll(100, TimeUnit.MILLISECONDS); + assertNotNull(result); + } + + @Test + public void testSize() throws Exception { + MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), "FOO"); + queue.put(new StringMessage("foo")); + assertEquals(1, queue.size()); + queue.poll(100, TimeUnit.MILLISECONDS); + assertEquals(0, queue.size()); + } + + @Test + public void testCapacityAfterExpiry() throws Exception { + SimpleMessageStore messageGroupStore = new SimpleMessageStore(); + MessageGroupQueue queue = new MessageGroupQueue(messageGroupStore, "FOO", 2); + queue.put(new StringMessage("foo")); + assertEquals(1, queue.remainingCapacity()); + queue.put(new StringMessage("bar")); + assertEquals(0, queue.remainingCapacity()); + Message result = queue.poll(100, TimeUnit.MILLISECONDS); + assertNotNull(result); + assertEquals(1, queue.remainingCapacity()); + } + + @Test + public void testCapacityExceeded() throws Exception { + SimpleMessageStore messageGroupStore = new SimpleMessageStore(); + MessageGroupQueue queue = new MessageGroupQueue(messageGroupStore, "FOO", 1); + queue.put(new StringMessage("foo")); + assertFalse(queue.offer(new StringMessage("bar"), 100, TimeUnit.MILLISECONDS)); + } + + @Test + public void testPutAndTake() throws Exception { + MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), "FOO"); + queue.put(new StringMessage("foo")); + Message result = queue.take(); + assertNotNull(result); + } + + @Test + public void testConcurrentAccess() throws Exception { + + SimpleMessageStore messageGroupStore = new SimpleMessageStore(); + final MessageGroupQueue queue = new MessageGroupQueue(messageGroupStore, "FOO"); + CompletionService completionService = new ExecutorCompletionService(Executors + .newCachedThreadPool()); + + int concurrency = 30; + final int maxPerTask = 20; + final Set set = new HashSet(); + + for (int i = 0; i < concurrency; i++) { + + final int big = i; + + completionService.submit(new Callable() { + public Boolean call() throws Exception { + boolean result = true; + for (int j = 0; j < maxPerTask; j++) { + result &= queue.add(new StringMessage("count=" + big + ":" + j)); + if (!result) { + logger.warn("Failed to add"); + } + } + return result; + } + }); + + completionService.submit(new Callable() { + public Boolean call() throws Exception { + boolean result = true; + for (int j = 0; j < maxPerTask; j++) { + @SuppressWarnings("unchecked") + Message item = (Message) queue.poll(1, TimeUnit.SECONDS); + set.add(item.getPayload()); + result &= item!=null; + if (!result) { + logger.warn("Failed to poll"); + } + } + return result; + } + }); + + messageGroupStore.expireMessageGroups(-10000); + + } + + for (int j = 0; j < 2*concurrency; j++) { + assertTrue(completionService.take().get()); + } + + // Ensure all items polled are unique + assertEquals(concurrency*maxPerTask, set.size()); + + assertEquals(0, queue.size()); + messageGroupStore.expireMessageGroups(-10000); + assertEquals(Integer.MAX_VALUE, queue.remainingCapacity()); + + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java index 75f88b05aa..1cca6c6e40 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java @@ -13,7 +13,7 @@ package org.springframework.integration.store; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; import java.util.ArrayList; import java.util.List; @@ -57,8 +57,9 @@ public class MessageStoreReaperTests { private static final List groups = new ArrayList(); - public void execute(MessageGroup group) { + public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { groups.add(group); + messageGroupStore.removeMessageGroup(group.getCorrelationKey()); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java index 8da6e76cea..6bcf2a75e6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java @@ -39,7 +39,7 @@ public class MessageStoreTests { public void shouldRegisterCallbacks() throws Exception { TestMessageStore store = new TestMessageStore(); store.setExpiryCallbacks(Arrays.asList(new MessageGroupCallback() { - public void execute(MessageGroup group) { + public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { } })); assertEquals(1, ((Collection)ReflectionTestUtils.getField(store, "expiryCallbacks")).size()); @@ -51,8 +51,9 @@ public class MessageStoreTests { TestMessageStore store = new TestMessageStore(); final List list = new ArrayList(); store.registerMessageGroupExpiryCallback(new MessageGroupCallback() { - public void execute(MessageGroup group) { + public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { list.add(group.getOne().getPayload().toString()); + messageGroupStore.removeMessageGroup(group.getCorrelationKey()); } }); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java index 265c22ccb1..fd53672c7b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java @@ -105,7 +105,7 @@ public class SimpleMessageStoreTests { public void shouldRegisterCallbacks() throws Exception { SimpleMessageStore store = new SimpleMessageStore(); store.setExpiryCallbacks(Arrays.asList(new MessageGroupCallback() { - public void execute(MessageGroup group) { + public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { } })); assertEquals(1, ((Collection)ReflectionTestUtils.getField(store, "expiryCallbacks")).size()); @@ -117,8 +117,9 @@ public class SimpleMessageStoreTests { SimpleMessageStore store = new SimpleMessageStore(); final List list = new ArrayList(); store.registerMessageGroupExpiryCallback(new MessageGroupCallback() { - public void execute(MessageGroup group) { + public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { list.add(group.getOne().getPayload().toString()); + messageGroupStore.removeMessageGroup(group.getCorrelationKey()); } }); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index a9b971f85c..10d00a8b0b 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -68,7 +68,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa 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_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 order by CREATED_DATE"; 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"; @@ -301,17 +301,15 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } public MessageGroup removeMessageFromGroup(Object correlationKey, Message messageToMark) { - final long updatedDate = System.currentTimeMillis(); final String correlationId = getKey(correlationKey); final String messageId = getKey(messageToMark.getHeaders().getId()); jdbcTemplate.update(getQuery(REMOVE_MESSAGE_FROM_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); - ps.setString(4, messageId); + logger.debug("Removing message from group with correlation key=" + correlationId); + ps.setString(1, correlationId); + ps.setString(2, region); + ps.setString(3, messageId); } }); return getMessageGroup(correlationKey); diff --git a/spring-integration-jdbc/src/main/resources/log4j.properties b/spring-integration-jdbc/src/test/java/log4j.properties similarity index 76% rename from spring-integration-jdbc/src/main/resources/log4j.properties rename to spring-integration-jdbc/src/test/java/log4j.properties index 54815027cf..a855018c87 100644 --- a/spring-integration-jdbc/src/main/resources/log4j.properties +++ b/spring-integration-jdbc/src/test/java/log4j.properties @@ -6,6 +6,6 @@ log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m log4j.category.org.springframework=WARN -log4j.category.org.springframework.integration=WARN +log4j.category.org.springframework.integration=DEBUG log4j.category.org.springframework.integration.jdbc=WARN -log4j.category.org.springframework.jdbc=WARN +log4j.category.org.springframework.jdbc=DEBUG diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests-context.xml new file mode 100644 index 0000000000..f7f4c8af7f --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests-context.xml @@ -0,0 +1,49 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests.java new file mode 100644 index 0000000000..f5450f9cc8 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests.java @@ -0,0 +1,95 @@ +package org.springframework.integration.jdbc; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.message.StringMessage; +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; + +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class JdbcMessageStoreChannelTests { + + @Autowired + private MessageChannel input; + + @Autowired + private JdbcMessageStore messageStore; + + @Before + public void clear() { + for (MessageGroup group : messageStore) { + messageStore.removeMessageGroup(group.getCorrelationKey()); + } + } + + @Test + public void testSendAndActivate() throws Exception { + Service.reset(1); + input.send(new StringMessage("foo")); + Service.await(1000); + assertEquals(1, Service.messages.size()); + assertEquals(0, messageStore.getMessageGroup("input-queue").size()); + } + + @Test + public void testSendAndActivateWithRollback() throws Exception { + Service.reset(1); + Service.fail = true; + input.send(new StringMessage("foo")); + Service.await(1000); + assertEquals(1, Service.messages.size()); + // After a rollback in the poller the message is still waiting to be delivered + assertEquals(1, messageStore.getMessageGroup("input-queue").size()); + assertEquals(1, messageStore.getMessageGroup("input-queue").getUnmarked().size()); + } + + @Test + @Transactional + public void testSendAndActivateTransactionalSend() throws Exception { + Service.reset(1); + input.send(new StringMessage("foo")); + // This will time out because the transaction has not committed yet + Service.await(1000); + // So no activation + assertEquals(0, Service.messages.size()); + // But inside the transaction the message is still there + assertEquals(1, messageStore.getMessageGroup("input-queue").size()); + assertEquals(1, messageStore.getMessageGroup("input-queue").getUnmarked().size()); + } + + public static class Service { + private static boolean fail = false; + private static List messages = new ArrayList(); + private static CountDownLatch latch = new CountDownLatch(0); + public static void reset(int count) { + fail = false; + messages.clear(); + latch = new CountDownLatch(count); + } + public static void await(long timeout) throws InterruptedException { + latch.await(timeout, TimeUnit.MILLISECONDS); + } + public String echo(String input) { + latch.countDown(); + messages.add(input); + if (fail) { + throw new RuntimeException("Planned failure"); + } + return input; + } + } + +} diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests-context.xml index 5adbde114a..f154c7042f 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests-context.xml @@ -2,7 +2,7 @@ diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index c5a0a7b984..0c58556e11 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -8,6 +8,7 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders; +import java.util.Iterator; import java.util.UUID; import javax.sql.DataSource; @@ -19,6 +20,8 @@ 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.integration.store.MessageGroupCallback; +import org.springframework.integration.store.MessageGroupStore; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.transaction.annotation.Transactional; @@ -100,7 +103,7 @@ public class JdbcMessageStoreTests { @Test @Transactional - public void testAddAndDelete() throws Exception { + public void testAddAndRemoveMessageGroup() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); message = messageStore.addMessage(message); assertNotNull(messageStore.removeMessage(message.getHeaders().getId())); @@ -118,6 +121,32 @@ public class JdbcMessageStoreTests { assertTrue("Timestamp too early: " + group.getTimestamp() + "<" + now, group.getTimestamp() >= now); } + @Test + @Transactional + public void testAddAndRemoveMessageFromMessageGroup() throws Exception { + String correlationId = "X"; + Message message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build(); + messageStore.addMessageToGroup(correlationId, message); + messageStore.removeMessageFromGroup(correlationId, message); + MessageGroup group = messageStore.getMessageGroup(correlationId); + assertEquals(0, group.size()); + } + + @Test + @Transactional + public void testOrderInMessageGroup() throws Exception { + String correlationId = "X"; + Message message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build(); + messageStore.addMessageToGroup(correlationId, message); + message = MessageBuilder.withPayload("bar").setCorrelationId(correlationId).build(); + messageStore.addMessageToGroup(correlationId, message); + MessageGroup group = messageStore.getMessageGroup(correlationId); + assertEquals(2, group.size()); + Iterator> iterator = group.getUnmarked().iterator(); + assertEquals("foo", iterator.next().getPayload()); + assertEquals("bar", iterator.next().getPayload()); + } + @Test @Transactional public void testAddAndMarkMessageGroup() throws Exception { @@ -135,9 +164,14 @@ public class JdbcMessageStoreTests { String correlationId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build(); messageStore.addMessageToGroup(correlationId, message); + messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() { + public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { + messageGroupStore.removeMessageGroup(group.getCorrelationKey()); + } + }); messageStore.expireMessageGroups(-10000); MessageGroup group = messageStore.getMessageGroup(correlationId); - assertEquals(0, group.getMarked().size()); + assertEquals(0, group.size()); } }