From 86c574e4935a9fe2e328ecd77eea58cf30387339 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 16 Dec 2011 09:53:39 -0500 Subject: [PATCH] INT-2311 modified implementation of MessageGroupQueue to use interruptible locks Basically modeled it after LinkedBlockingQueue changed to use ReadWriteLock refactored MessageGroupQueue and added tests changed MessageGroupQueue to use Condition added messageGroupSize(groupId) method to MessageGroupStore and all the implementations polishing based on Gary's comments removed setStoreLock method in favor of additional constructors polishing tests, decreased timeouts made peek() thread safe added logging --- .../store/AbstractKeyValueMessageStore.java | 11 +- .../store/MessageGroupMetadata.java | 4 + .../integration/store/MessageGroupQueue.java | 308 +++++++++++------ .../integration/store/MessageGroupStore.java | 7 + .../integration/store/SimpleMessageStore.java | 4 + .../integration/store/MessageStoreTests.java | 18 +- .../integration/jdbc/JdbcMessageStore.java | 8 +- ...geStoreChannelIntegrationTests-context.xml | 2 +- .../integration/jdbc/LockInterceptor.java | 6 +- .../jdbc/MessageGroupQueueTests.java | 318 ++++++++++++++++++ .../mongodb/store/MongoDbMessageStore.java | 6 + .../store/MongoDbMessageGroupStoreTests.java | 13 + 12 files changed, 580 insertions(+), 125 deletions(-) create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/MessageGroupQueueTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java index 803ee4e96a..10902b69f6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java @@ -203,6 +203,16 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS return new MessageGroupIterator(idIterator); } + public int messageGroupSize(Object groupId) { + Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); + if (mgm != null) { + Assert.isInstanceOf(MessageGroupMetadata.class, mgm); + MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; + return messageGroupMetadata.size(); + } + return 0; + } + protected abstract Object doRetrieve(Object id); protected abstract void doStore(Object id, Object objectToStore); @@ -308,5 +318,4 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS throw new UnsupportedOperationException(); } } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java index b99937fce1..2d1445afc0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java @@ -76,6 +76,10 @@ public class MessageGroupMetadata implements Serializable{ return this.messageIds.iterator(); } + public int size(){ + return this.messageIds.size(); + } + public UUID firstId(){ if (this.messageIds.size() > 0){ return this.messageIds.iterator().next(); 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 index ffa664db28..d4c3473492 100644 --- 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 @@ -21,8 +21,15 @@ import java.util.Collection; import java.util.Iterator; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.integration.Message; +import org.springframework.util.Assert; /** * A {@link BlockingQueue} that is backed by a {@link MessageGroupStore}. Can be used to ensure guaranteed delivery in @@ -31,159 +38,223 @@ import org.springframework.integration.Message; * must be provided, so it needs to be unique but identifiable with a single logical instance of the queue. * * @author Dave Syer + * @author Oleg Zhurakousky * @since 2.0 * */ public class MessageGroupQueue extends AbstractQueue> implements BlockingQueue> { + + private final Log logger = LogFactory.getLog(getClass()); - private static final int DEFAULT_CAPACITY = -1; + private static final int DEFAULT_CAPACITY = Integer.MAX_VALUE; private final MessageGroupStore messageGroupStore; private final Object groupId; private final int capacity; - - // This one could be a global semaphore - private volatile Object storeLock = new Object(); - - // This one only needs to be local - private final Object writeLock = new Object(); - - // This one only needs to be local - private final Object readLock = new Object(); - + + //This one could be a global semaphore + private final Lock storeLock; + + private final Condition messageStoreNotFull; + + private final Condition messageStoreNotEmpty; + public MessageGroupQueue(MessageGroupStore messageGroupStore, Object groupId) { - this(messageGroupStore, groupId, DEFAULT_CAPACITY); + this(messageGroupStore, groupId, DEFAULT_CAPACITY, new ReentrantLock(true)); } public MessageGroupQueue(MessageGroupStore messageGroupStore, Object groupId, int capacity) { + this(messageGroupStore, groupId, capacity, new ReentrantLock(true)); + } + + public MessageGroupQueue(MessageGroupStore messageGroupStore, Object groupId, Lock storeLock) { + this(messageGroupStore, groupId, DEFAULT_CAPACITY, storeLock); + } + + public MessageGroupQueue(MessageGroupStore messageGroupStore, Object groupId, int capacity, Lock storeLock) { + Assert.isTrue(capacity > 0, "'capacity' must be greater than 0"); + Assert.notNull(storeLock, "'storeLock' must not be null"); + Assert.notNull(messageGroupStore, "'messageGroupStore' must not be null"); + Assert.notNull(groupId, "'groupId' must not be null"); + this.storeLock = storeLock; + this.messageStoreNotFull = this.storeLock.newCondition(); + this.messageStoreNotEmpty = this.storeLock.newCondition(); this.messageGroupStore = messageGroupStore; this.groupId = groupId; this.capacity = capacity; } - - /** - * @param storeLock the storeLock to set - */ - public void setStoreLock(Object storeLock) { - this.storeLock = storeLock; - } public Iterator> iterator() { return getMessages().iterator(); } public int size() { - return this.messageGroupStore.getMessageGroup(groupId).size(); - } - - public boolean offer(Message e) { - synchronized (storeLock) { - if (capacity>0 && messageGroupStore.getMessageGroup(groupId).size() >= capacity) { - return false; - } - messageGroupStore.addMessageToGroup(groupId, e); - } - synchronized (readLock) { - readLock.notifyAll(); - } - return true; + return messageGroupStore.messageGroupSize(groupId); } public Message peek() { - Collection> messages = getMessages(); - if (messages.isEmpty()) { - return null; - } - return messages.iterator().next(); - } - - public Message poll() { - Message result = null; - synchronized (storeLock) { - result = this.messageGroupStore.pollMessageFromGroup(groupId); - } - synchronized (writeLock) { - writeLock.notifyAll(); - } - return result; - } - - public int drainTo(Collection> c) { - synchronized (storeLock) { - for (Message message = this.messageGroupStore.pollMessageFromGroup(groupId); message != null;) { - c.add(message); + Message message = null; + final Lock storeLock = this.storeLock; + try { + storeLock.lockInterruptibly(); + try { + Collection> messages = getMessages(); + if (!messages.isEmpty()) { + message = messages.iterator().next(); + } + } + finally { + storeLock.unlock(); } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); } - synchronized (writeLock) { - writeLock.notifyAll(); - } - return this.messageGroupStore.getMessageGroup(groupId).size(); + return message; } - - public int drainTo(Collection> c, int maxElements) { - ArrayList> list = new ArrayList>(); - synchronized (storeLock) { - Message message = this.messageGroupStore.pollMessageFromGroup(groupId); - for (int i = 0; i < maxElements && message != null; i++) { - list.add(message); - message = this.messageGroupStore.pollMessageFromGroup(groupId); - } - } - synchronized (writeLock) { - writeLock.notifyAll(); - } - c.addAll(list); - return list.size(); - } - - public boolean offer(Message e, long timeout, TimeUnit unit) throws InterruptedException { - long threshold = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(timeout, unit); - boolean result = offer(e); - while (!result && System.currentTimeMillis() < threshold) { - synchronized (writeLock) { - writeLock.wait(threshold - System.currentTimeMillis()); - } - result = offer(e); - } - return result; - } - + 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 message = null; + long timeoutInNanos = unit.toNanos(timeout); + final Lock storeLock = this.storeLock; + storeLock.lockInterruptibly(); + + try { + while (this.size() == 0 && timeoutInNanos > 0){ + timeoutInNanos = this.messageStoreNotEmpty.awaitNanos(timeoutInNanos); } - message = poll(); + message = this.doPoll(); + + } + finally { + storeLock.unlock(); } return message; } - public void put(Message e) throws InterruptedException { - while (!offer(e)) { - synchronized (writeLock) { - writeLock.wait(); + public Message poll() { + Message message = null; + final Lock storeLock = this.storeLock; + try { + storeLock.lockInterruptibly(); + try { + message = this.doPoll(); + } + finally { + storeLock.unlock(); } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return message; + } + + public int drainTo(Collection> c) { + return this.drainTo(c, Integer.MAX_VALUE); + } + + public int drainTo(Collection> collection, int maxElements) { + Assert.notNull(collection, "'collection' must not be null"); + int originalSize = collection.size(); + ArrayList> list = new ArrayList>(); + final Lock storeLock = this.storeLock; + try { + storeLock.lockInterruptibly(); + try { + Message message = this.messageGroupStore.pollMessageFromGroup(groupId); + for (int i = 0; i < maxElements && message != null; i++) { + list.add(message); + message = this.messageGroupStore.pollMessageFromGroup(groupId); + } + this.messageStoreNotFull.signal(); + } + finally { + storeLock.unlock(); + } + } + catch (InterruptedException e) { + logger.warn("Queue may not have drained completely since this operation was interrupted", e); + Thread.currentThread().interrupt(); + } + collection.addAll(list); + return collection.size() - originalSize; + } + + public boolean offer(Message message) { + boolean offered = true; + final Lock storeLock = this.storeLock; + try { + storeLock.lockInterruptibly(); + try { + offered = this.doOffer(message); + } + finally { + storeLock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return offered; + } + + public boolean offer(Message message, long timeout, TimeUnit unit) throws InterruptedException { + long timeoutInNanos = unit.toNanos(timeout); + boolean offered = false; + + final Lock storeLock = this.storeLock; + storeLock.lockInterruptibly(); + try { + while (this.size() == capacity && timeoutInNanos > 0){ + timeoutInNanos = this.messageStoreNotFull.awaitNanos(timeoutInNanos); + } + + if (timeoutInNanos > 0){ + offered = this.doOffer(message); + } + } + finally { + storeLock.unlock(); + } + return offered; + } + + public void put(Message message) throws InterruptedException { + final Lock storeLock = this.storeLock; + storeLock.lockInterruptibly(); + try { + while (this.size() == capacity){ + this.messageStoreNotFull.await(); + } + + this.doOffer(message); + } + finally { + storeLock.unlock(); } } public int remainingCapacity() { - return (capacity>0 ? capacity : Integer.MAX_VALUE) - messageGroupStore.getMessageGroup(groupId).size(); + return capacity - this.size(); } public Message take() throws InterruptedException { - Message message = poll(); - while (message == null) { - synchronized (readLock) { - readLock.wait(); + Message message = null; + final Lock storeLock = this.storeLock; + storeLock.lockInterruptibly(); + + try { + while (this.size() == 0){ + this.messageStoreNotEmpty.await(); } - message = poll(); + message = this.doPoll(); + + } + finally { + storeLock.unlock(); } return message; } @@ -192,4 +263,27 @@ public class MessageGroupQueue extends AbstractQueue> implements Bloc return messageGroupStore.getMessageGroup(groupId).getMessages(); } + /** + * It is assumed that the 'storeLock' is being held by the caller, otherwise + * IllegalMonitorStateException may be thrown + */ + private Message doPoll() { + Message message = this.messageGroupStore.pollMessageFromGroup(groupId); + this.messageStoreNotFull.signal(); + return message; + } + + /** + * It is assumed that the 'storeLock' is being held by the caller, otherwise + * IllegalMonitorStateException may be thrown + */ + private boolean doOffer(Message message){ + boolean offered = false; + if (this.size() < capacity){ + messageGroupStore.addMessageToGroup(groupId, message); + offered = true; + this.messageStoreNotEmpty.signal(); + } + return offered; + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java index cc988485b7..2a04cf437b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java @@ -46,6 +46,13 @@ public interface MessageGroupStore { */ @ManagedAttribute int getMessageGroupCount(); + + /** + * Returns the size of this MessageGroup + * @param groupId + */ + @ManagedAttribute + int messageGroupSize(Object groupId); /** * Return all Messages currently in the MessageStore that were stored using diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index 998e522699..3588908c25 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -202,4 +202,8 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes } return lock; } + + public int messageGroupSize(Object groupId) { + return this.getMessageGroup(groupId).size(); + } } 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 0b5536c5af..1f29a4e280 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 @@ -16,8 +16,6 @@ package org.springframework.integration.store; -import static org.junit.Assert.assertEquals; - import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -25,10 +23,13 @@ import java.util.Iterator; import java.util.List; import org.junit.Test; + import org.springframework.integration.Message; import org.springframework.integration.message.GenericMessage; import org.springframework.test.util.ReflectionTestUtils; +import static org.junit.Assert.assertEquals; + /** * @author Dave Syer */ @@ -94,18 +95,10 @@ public class MessageStoreTests { return removed ? new SimpleMessageGroup(correlationKey) : testMessages; } - public MessageGroup markMessageGroup(MessageGroup group) { - throw new UnsupportedOperationException(); - } - public MessageGroup removeMessageFromGroup(Object key, Message messageToRemove) { throw new UnsupportedOperationException(); } - public MessageGroup markMessageFromGroup(Object key, Message messageToMark) { - throw new UnsupportedOperationException(); - } - public void removeMessageGroup(Object correlationKey) { if (correlationKey.equals(testMessages.getGroupId())) { removed = true; @@ -122,10 +115,13 @@ public class MessageStoreTests { } public Message pollMessageFromGroup(Object groupId) { - // TODO Auto-generated method stub return null; } + public int messageGroupSize(Object groupId) { + return 0; + } + } } 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 551e7c5b14..ae82a0f55a 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 @@ -95,7 +95,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private static final String COUNT_ALL_GROUPS = "SELECT COUNT(GROUP_KEY) from %PREFIX%MESSAGE_GROUP where REGION=?"; - private static final String COUNT_ALL_MARKED_MESSAGES_IN_GROUPS = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE_GROUP where MARKED=1 AND REGION=?"; + private static final String COUNT_ALL_MESSAGES_IN_GROUP = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? AND REGION=?"; private static final String COUNT_ALL_MESSAGES_IN_GROUPS = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE_GROUP where REGION=?"; @@ -348,8 +348,9 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } @ManagedAttribute - public int getMarkedMessageCountForAllMessageGroups() { - return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_MARKED_MESSAGES_IN_GROUPS), region); + public int messageGroupSize(Object groupId) { + String key = getKey(groupId); + return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_MESSAGES_IN_GROUP), key, region); } public MessageGroup getMessageGroup(Object groupId) { @@ -467,7 +468,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa public Message pollMessageFromGroup(final Object groupId) { String key = getKey(groupId); - Message message = jdbcTemplate.query(getQuery(LIST_MESSAGEIDS_BY_GROUP_KEY), new Object[] { key, region }, new ResultSetExtractor>() { public Message extractData(ResultSet rs) diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml index 289a5b802e..2f60e04b60 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml @@ -28,7 +28,7 @@ - + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/LockInterceptor.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/LockInterceptor.java index 1d961413c4..65de5e9106 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/LockInterceptor.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/LockInterceptor.java @@ -13,6 +13,8 @@ package org.springframework.integration.jdbc; +import java.util.concurrent.locks.ReentrantLock; + import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; @@ -20,7 +22,9 @@ import org.aopalliance.intercept.MethodInvocation; * @author Dave Syer * */ -public class LockInterceptor implements MethodInterceptor { +public class LockInterceptor extends ReentrantLock implements MethodInterceptor { + + private static final long serialVersionUID = 1L; public synchronized Object invoke(MethodInvocation invocation) throws Throwable { return invocation.proceed(); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/MessageGroupQueueTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/MessageGroupQueueTests.java new file mode 100644 index 0000000000..9559b86966 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/MessageGroupQueueTests.java @@ -0,0 +1,318 @@ +/* + * Copyright 2002-2011 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.jdbc; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Test; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.springframework.integration.Message; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageGroupQueue; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.SimpleMessageStore; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +/** + * @author Oleg Zhurakousky + */ +public class MessageGroupQueueTests { + + + @Test + public void validateMgqInterruption() throws Exception{ + + final MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), 1, 1); + + final AtomicReference exceptionHolder = new AtomicReference(); + + Thread t = new Thread(new Runnable() { + + public void run() { + queue.offer(new GenericMessage("hello")); + try { + queue.offer(new GenericMessage("hello"), 100, TimeUnit.SECONDS); + } catch (InterruptedException e) { + exceptionHolder.set(e); + } + } + }); + t.start(); + Thread.sleep(1000); + t.interrupt(); + Thread.sleep(1000); + assertTrue(exceptionHolder.get() instanceof InterruptedException); + } + + @Test + public void testConcurrentReadWrite() throws Exception{ + final MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), 1, 1); + final AtomicReference> messageHolder = new AtomicReference>(); + + Thread t1 = new Thread(new Runnable() { + public void run() { + try { + messageHolder.set(queue.poll(1000, TimeUnit.SECONDS)); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + Thread t2 = new Thread(new Runnable() { + public void run() { + try { + queue.offer(new GenericMessage("hello"), 1000, TimeUnit.SECONDS); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t1.start(); + t2.start(); + Thread.sleep(1000); + assertTrue(messageHolder.get() instanceof Message); + } + + @Test + public void testConcurrentWriteRead() throws Exception{ + final MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), 1, 1); + final AtomicReference> messageHolder = new AtomicReference>(); + + queue.offer(new GenericMessage("hello"), 1000, TimeUnit.SECONDS); + + Thread t1 = new Thread(new Runnable() { + public void run() { + try { + queue.offer(new GenericMessage("Hi"), 1000, TimeUnit.SECONDS); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + Thread t2 = new Thread(new Runnable() { + public void run() { + try { + queue.poll(1000, TimeUnit.SECONDS); + messageHolder.set(queue.poll(1000, TimeUnit.SECONDS)); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + + t1.start(); + Thread.sleep(1000); + t2.start(); + Thread.sleep(1000); + assertTrue(messageHolder.get().getPayload().equals("Hi")); + } + + @Test + public void testConcurrentReadersWithTimeout() throws Exception{ + final MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), 1, 1); + final AtomicReference> messageHolder1 = new AtomicReference>(); + final AtomicReference> messageHolder2 = new AtomicReference>(); + final AtomicReference> messageHolder3 = new AtomicReference>(); + + Thread t1 = new Thread(new Runnable() { + public void run() { + try { + messageHolder1.set(queue.poll(10, TimeUnit.SECONDS)); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + Thread t2 = new Thread(new Runnable() { + public void run() { + try { + messageHolder2.set(queue.poll(10, TimeUnit.SECONDS)); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + Thread t3 = new Thread(new Runnable() { + public void run() { + try { + messageHolder3.set(queue.poll(10, TimeUnit.SECONDS)); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + Thread t4 = new Thread(new Runnable() { + public void run() { + try { + queue.offer(new GenericMessage("Hi"), 10, TimeUnit.SECONDS); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t1.start(); + Thread.sleep(1000); + t2.start(); + Thread.sleep(1000); + t3.start(); + Thread.sleep(1000); + t4.start(); + Thread.sleep(1000); + assertTrue(messageHolder1.get().getPayload().equals("Hi")); + Thread.sleep(4000); + assertTrue(messageHolder2.get() == null); + } + + @Test + public void testConcurrentWritersWithTimeout() throws Exception{ + final MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), 1, 1); + final AtomicReference booleanHolder1 = new AtomicReference(true); + final AtomicReference booleanHolder2 = new AtomicReference(true); + final AtomicReference booleanHolder3 = new AtomicReference(true); + + Thread t1 = new Thread(new Runnable() { + public void run() { + try { + booleanHolder1.set(queue.offer(new GenericMessage("Hi-1"), 2, TimeUnit.SECONDS)); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + Thread t2 = new Thread(new Runnable() { + public void run() { + try { + boolean offered = queue.offer(new GenericMessage("Hi-2"), 2, TimeUnit.SECONDS); + System.out.println(offered); + booleanHolder2.set(offered); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + Thread t3 = new Thread(new Runnable() { + public void run() { + try { + boolean offered = queue.offer(new GenericMessage("Hi-3"), 2, TimeUnit.SECONDS); + System.out.println(offered); + booleanHolder3.set(offered); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t1.start(); + Thread.sleep(1000); + t2.start(); + Thread.sleep(100); + t3.start(); + Thread.sleep(4000); + assertTrue(booleanHolder1.get()); + assertFalse(booleanHolder2.get()); + assertFalse(booleanHolder3.get()); + } + @Test + public void testConcurrentWriteReadMulti() throws Exception{ + final MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), 1, 4); + final AtomicReference> messageHolder = new AtomicReference>(); + + queue.offer(new GenericMessage("hello"), 1000, TimeUnit.SECONDS); + + Thread t1 = new Thread(new Runnable() { + public void run() { + try { + queue.offer(new GenericMessage("Hi"), 1000, TimeUnit.SECONDS); + queue.offer(new GenericMessage("Hi"), 1000, TimeUnit.SECONDS); + queue.offer(new GenericMessage("Hi"), 1000, TimeUnit.SECONDS); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + Thread t2 = new Thread(new Runnable() { + public void run() { + try { + queue.poll(1000, TimeUnit.SECONDS); + messageHolder.set(queue.poll(1000, TimeUnit.SECONDS)); + queue.poll(1000, TimeUnit.SECONDS); + queue.poll(1000, TimeUnit.SECONDS); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + + t1.start(); + Thread.sleep(1000); + t2.start(); + Thread.sleep(1000); + assertTrue(messageHolder.get().getPayload().equals("Hi")); + assertNull(queue.poll(5, TimeUnit.SECONDS)); + } + + @Test + public void validateMgqInterruptionStoreLock() throws Exception{ + + MessageGroupStore mgs = Mockito.mock(MessageGroupStore.class); + Mockito.doAnswer(new Answer() { + public MessageGroup answer(InvocationOnMock invocation) + throws Throwable { + Thread.sleep(5000); + return null; + } + }).when(mgs).addMessageToGroup(Mockito.any(Integer.class), Mockito.any(Message.class)); + + MessageGroup mg = Mockito.mock(MessageGroup.class); + Mockito.when(mgs.getMessageGroup(Mockito.any())).thenReturn(mg); + Mockito.when(mg.size()).thenReturn(0); + + final MessageGroupQueue queue = new MessageGroupQueue(mgs, 1, 1); + + final AtomicReference exceptionHolder = new AtomicReference(); + + Thread t1 = new Thread(new Runnable() { + + public void run() { + queue.offer(new GenericMessage("hello")); + } + }); + t1.start(); + Thread.sleep(500); + Thread t2 = new Thread(new Runnable() { + + public void run() { + queue.offer(new GenericMessage("hello")); + try { + queue.offer(new GenericMessage("hello"), 100, TimeUnit.SECONDS); + } catch (InterruptedException e) { + exceptionHolder.set(e); + } + } + }); + t2.start(); + Thread.sleep(1000); + t2.interrupt(); + Thread.sleep(1000); + assertTrue(exceptionHolder.get() instanceof InterruptedException); + } +} diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java index bf4d26defa..274e3fbf95 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java @@ -245,6 +245,12 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me this.updateGroup(groupId); return message; } + + public int messageGroupSize(Object groupId) { + long lCount = this.template.count(new Query(where(GROUP_ID_KEY).is(groupId)), this.collectionName); + Assert.isTrue(lCount <= Integer.MAX_VALUE, "Message count is out of Integer's range"); + return (int) lCount; + } /* * Common Queries diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java index 7306da7906..cb81f41be3 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java @@ -75,6 +75,19 @@ public class MongoDbMessageGroupStoreTests extends MongoDbAvailableTests { assertNull(retrievedMessage.getHeaders().get("message_group")); } + @Test + @MongoDbAvailable + public void testCountMessagesInGroup() throws Exception{ + MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); + MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); + + Message messageA = new GenericMessage("A"); + Message messageB = new GenericMessage("B"); + store.addMessageToGroup(1, messageA); + store.addMessageToGroup(1, messageB); + assertEquals(2, store.messageGroupSize(1)); + } + @Test @MongoDbAvailable public void testMessageGroupUpdatedDateChangesWithEachAddedMessage() throws Exception{