Merge pull request #248 from olegz/INT-2311

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();
 		}
 	}
-
 }

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();

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<Message<?>> implements BlockingQueue<Message<?>> {
+	
+	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<Message<?>> 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<Message<?>> 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<? super Message<?>> 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<Message<?>> 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<? super Message<?>> c, int maxElements) {
-		ArrayList<Message<?>> list = new ArrayList<Message<?>>();
-		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<? super Message<?>> c) {
+		return this.drainTo(c, Integer.MAX_VALUE);
+	}
+
+	public int drainTo(Collection<? super Message<?>> collection, int maxElements) {
+		Assert.notNull(collection, "'collection' must not be null");
+		int originalSize = collection.size();
+		ArrayList<Message<?>> list = new ArrayList<Message<?>>();
+		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<Message<?>> 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;
+	}
 }

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

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();
+	}
 }

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;
+		}
+
 	}
 
 }

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<Message<?>>() {
 			public Message<?> extractData(ResultSet rs)

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml

@@ -28,7 +28,7 @@
 	<bean id="storeQueue" class="org.springframework.integration.store.MessageGroupQueue">
 		<constructor-arg ref="messageStore" />
 		<constructor-arg value="JdbcMessageStoreChannelIntegrationTests" />
-		<property name="storeLock" ref="lock" />
+		<constructor-arg ref="lock" />
 	</bean>
 
 	<int:channel id="output" />

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();

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<InterruptedException> exceptionHolder = new AtomicReference<InterruptedException>();
+		
+		Thread t = new Thread(new Runnable() {
+			
+			public void run() {
+				queue.offer(new GenericMessage<String>("hello"));
+				try {
+					queue.offer(new GenericMessage<String>("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<Message<?>> messageHolder = new AtomicReference<Message<?>>();
+		
+		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<String>("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<Message<?>> messageHolder = new AtomicReference<Message<?>>();
+		
+		queue.offer(new GenericMessage<String>("hello"), 1000, TimeUnit.SECONDS);
+		
+		Thread t1 = new Thread(new Runnable() {		
+			public void run() {
+				try {
+					queue.offer(new GenericMessage<String>("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<Message<?>> messageHolder1 = new AtomicReference<Message<?>>();
+		final AtomicReference<Message<?>> messageHolder2 = new AtomicReference<Message<?>>();
+		final AtomicReference<Message<?>> messageHolder3 = new AtomicReference<Message<?>>();
+		
+		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<String>("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<Boolean> booleanHolder1 = new AtomicReference<Boolean>(true);
+		final AtomicReference<Boolean> booleanHolder2 = new AtomicReference<Boolean>(true);
+		final AtomicReference<Boolean> booleanHolder3 = new AtomicReference<Boolean>(true);
+		
+		Thread t1 = new Thread(new Runnable() {		
+			public void run() {
+				try {
+					booleanHolder1.set(queue.offer(new GenericMessage<String>("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<String>("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<String>("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<Message<?>> messageHolder = new AtomicReference<Message<?>>();
+            
+            queue.offer(new GenericMessage<String>("hello"), 1000, TimeUnit.SECONDS);
+            
+            Thread t1 = new Thread(new Runnable() {                
+                    public void run() {
+                            try {
+                                    queue.offer(new GenericMessage<String>("Hi"), 1000, TimeUnit.SECONDS);
+                                    queue.offer(new GenericMessage<String>("Hi"), 1000, TimeUnit.SECONDS);
+                                    queue.offer(new GenericMessage<String>("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<MessageGroup>() {
+                    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<InterruptedException> exceptionHolder = new AtomicReference<InterruptedException>();
+            
+            Thread t1 = new Thread(new Runnable() {
+                    
+                    public void run() {
+                            queue.offer(new GenericMessage<String>("hello"));
+                    }
+            });
+            t1.start();
+            Thread.sleep(500);
+            Thread t2 = new Thread(new Runnable() {
+                    
+                    public void run() {
+                            queue.offer(new GenericMessage<String>("hello"));
+                            try {
+                                    queue.offer(new GenericMessage<String>("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);
+    }
+}

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

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<String>("A");
+		Message<?> messageB = new GenericMessage<String>("B");
+		store.addMessageToGroup(1, messageA);
+		store.addMessageToGroup(1, messageB);
+		assertEquals(2, store.messageGroupSize(1));
+	}
+	
 	@Test
 	@MongoDbAvailable
 	public void testMessageGroupUpdatedDateChangesWithEachAddedMessage() throws Exception{