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{