diff options
19 files changed, 775 insertions, 416 deletions
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java index 5bde27dba5..8dac12fe24 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java @@ -60,30 +60,6 @@ public interface AMQMessage //Check the status of this message - /** - * Called selectors to determin if the message has already been sent - * - * @return _deliveredToConsumer - */ - boolean getDeliveredToConsumer(); - - /** - * Called to enforce the 'immediate' flag. - * - * @returns true if the message is marked for immediate delivery but has not been marked as delivered - * to a consumer - */ - boolean immediateAndNotDelivered(); - - /** - * Checks to see if the message has expired. If it has the message is dequeued. - * - * @return true if the message has expire - * - * @throws org.apache.qpid.AMQException - */ - boolean expired() throws AMQException; - /** Is this a persistent message * * @return true if the message is persistent @@ -91,13 +67,8 @@ public interface AMQMessage boolean isPersistent(); - /** - * Called when this message is delivered to a consumer. (used to implement the 'immediate' flag functionality). - * And for selector efficiency. - */ - void setDeliveredToConsumer(); + boolean isImmediate(); - void setExpiration(long expiration); void setClientIdentifier(AMQProtocolSession.ProtocolSessionIdentifier sessionIdentifier); @@ -121,4 +92,8 @@ public interface AMQMessage String toString(); String debugIdentity(); + + void setExpiration(long expiration); + + long getExpiration(); } diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQPriorityQueue.java b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQPriorityQueue.java index 34a70c6969..ade780d0bb 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQPriorityQueue.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQPriorityQueue.java @@ -36,12 +36,12 @@ public class AMQPriorityQueue extends SimpleAMQQueue int priorities) throws AMQException { - super(name, durable, owner, autoDelete, virtualHost, new PriorityQueueList.Factory(priorities)); + super(name, durable, owner, autoDelete, virtualHost, new PriorityQueueEntryList.Factory(priorities)); } public int getPriorities() { - return ((PriorityQueueList) _entries).getPriorities(); + return ((PriorityQueueEntryList) _entries).getPriorities(); } @Override diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/FlowableBaseQueueEntryList.java b/java/broker/src/main/java/org/apache/qpid/server/queue/FlowableBaseQueueEntryList.java new file mode 100644 index 0000000000..72ea5f2667 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/FlowableBaseQueueEntryList.java @@ -0,0 +1,185 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.qpid.server.queue; + +import org.apache.qpid.server.virtualhost.VirtualHost; +import org.apache.log4j.Logger; + +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +/** + * This is an abstract base class to handle + */ +public abstract class FlowableBaseQueueEntryList implements FlowableQueueEntryList +{ + private static final Logger _log = Logger.getLogger(FlowableBaseQueueEntryList.class); + + private final AtomicInteger _atomicQueueCount = new AtomicInteger(0); + private final AtomicLong _atomicQueueSize = new AtomicLong(0L); + private final AtomicLong _atomicQueueInMemory = new AtomicLong(0L); + /** The maximum amount of memory that is allocated to this queue. Beyond this the queue will flow to disk. */ + + private long _memoryUsageMaximum = 0; + + /** The minimum amount of memory that is allocated to this queue. If the queueDepth hits this level then more flowed data can be read in. */ + private long _memoryUsageMinimum = 0; + private AtomicBoolean _flowed; + private QueueBackingStore _backingStore; + protected AMQQueue _queue; + + FlowableBaseQueueEntryList(AMQQueue queue) + { + _queue = queue; + _flowed = new AtomicBoolean(false); + VirtualHost vhost = queue.getVirtualHost(); + if (vhost != null) + { + _backingStore = vhost.getQueueBackingStore(); + } + } + + public void setFlowed(boolean flowed) + { + if (_flowed.get() != flowed) + { + _log.info("Marking Queue(" + _queue.getName() + ") as flowed (" + flowed + ")"); + _flowed.set(flowed); + } + } + + public boolean isFlowed() + { + return _flowed.get(); + } + + public int size() + { + return _atomicQueueCount.get(); + } + + public long dataSize() + { + return _atomicQueueSize.get(); + } + + public long memoryUsed() + { + return _atomicQueueInMemory.get(); + } + + public void setMemoryUsageMaximum(long maximumMemoryUsage) + { + _memoryUsageMaximum = maximumMemoryUsage; + + // Don't attempt to start the inhaler/purger unless we have a minimum value specified. + if (_memoryUsageMaximum > 0) + { + // If we've increased the max memory above what we have in memory then we can inhale more + if (_memoryUsageMaximum > _atomicQueueInMemory.get()) + { + //TODO start inhaler + } + else // if we have now have to much memory in use we need to purge. + { + //TODO start purger + } + } + } + + public long getMemoryUsageMaximum() + { + return _memoryUsageMaximum; + } + + public void setMemoryUsageMinimum(long minimumMemoryUsage) + { + _memoryUsageMinimum = minimumMemoryUsage; + + // Don't attempt to start the inhaler unless we have a minimum value specified. + if (_memoryUsageMinimum > 0) + { + // If we've increased the minimum memory above what we have in memory then we need to inhale more + if (_memoryUsageMinimum >= _atomicQueueInMemory.get()) + { + //TODO start inhaler + } + } + } + + public long getMemoryUsageMinimum() + { + return _memoryUsageMinimum; + } + + protected boolean willCauseFlowToDisk(QueueEntryImpl queueEntry) + { + return _memoryUsageMaximum != 0 && memoryUsed() + queueEntry.getSize() > _memoryUsageMaximum; + } + + protected void incrementCounters(final QueueEntryImpl queueEntry) + { + _atomicQueueCount.incrementAndGet(); + _atomicQueueSize.addAndGet(queueEntry.getSize()); + if (!willCauseFlowToDisk(queueEntry)) + { + _atomicQueueInMemory.addAndGet(queueEntry.getSize()); + } + else + { + setFlowed(true); + flowingToDisk(queueEntry); + } + } + + /** + * Called when we are now flowing to disk + * @param queueEntry the entry that is being flowed to disk + */ + protected void flowingToDisk(QueueEntryImpl queueEntry) + { + try + { + queueEntry.flow(); + } + catch (UnableToFlowMessageException e) + { + _atomicQueueInMemory.addAndGet(queueEntry.getSize()); + } + } + + protected void dequeued(QueueEntryImpl queueEntry) + { + _atomicQueueCount.decrementAndGet(); + _atomicQueueSize.addAndGet(-queueEntry.getSize()); + if (!queueEntry.isFlowed()) + { + _atomicQueueInMemory.addAndGet(-queueEntry.getSize()); + } + } + + public QueueBackingStore getBackingStore() + { + return _backingStore; + } + +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/FlowableQueueEntryList.java b/java/broker/src/main/java/org/apache/qpid/server/queue/FlowableQueueEntryList.java new file mode 100644 index 0000000000..4e95978bf8 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/FlowableQueueEntryList.java @@ -0,0 +1,42 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.qpid.server.queue; + +public interface FlowableQueueEntryList +{ + void setFlowed(boolean flowed); + + boolean isFlowed(); + + int size(); + + long dataSize(); + + long memoryUsed(); + + void setMemoryUsageMaximum(long maximumMemoryUsage); + + long getMemoryUsageMaximum(); + + void setMemoryUsageMinimum(long minimumMemoryUsage); + + long getMemoryUsageMinimum(); +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/PriorityQueueList.java b/java/broker/src/main/java/org/apache/qpid/server/queue/PriorityQueueEntryList.java index 7be2827e0f..d812b8ceca 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/PriorityQueueList.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/PriorityQueueEntryList.java @@ -21,17 +21,17 @@ package org.apache.qpid.server.queue; import org.apache.qpid.framing.CommonContentHeaderProperties; -import org.apache.qpid.AMQException; -public class PriorityQueueList implements QueueEntryList +public class PriorityQueueEntryList extends FlowableBaseQueueEntryList implements QueueEntryList { private final AMQQueue _queue; private final QueueEntryList[] _priorityLists; private final int _priorities; private final int _priorityOffset; - public PriorityQueueList(AMQQueue queue, int priorities) + public PriorityQueueEntryList(AMQQueue queue, int priorities) { + super(queue); _queue = queue; _priorityLists = new QueueEntryList[priorities]; _priorities = priorities; @@ -53,7 +53,7 @@ public class PriorityQueueList implements QueueEntryList } public QueueEntry add(AMQMessage message) - { + { int index = ((CommonContentHeaderProperties)((message.getContentHeaderBody().properties))).getPriority() - _priorityOffset; if(index >= _priorities) { @@ -154,7 +154,29 @@ public class PriorityQueueList implements QueueEntryList public QueueEntryList createQueueEntryList(AMQQueue queue) { - return new PriorityQueueList(queue, _priorities); + return new PriorityQueueEntryList(queue, _priorities); + } + } + + @Override + public int size() + { + int size=0; + for (QueueEntryList queueEntryList : _priorityLists) + { + size += queueEntryList.size(); } + + return size; + } + + + @Override + protected void flowingToDisk(QueueEntryImpl queueEntry) + { + //TODO this disables FTD for priority queues + // As the incomming message isn't always the one to purge. + // More logic is required up in the add() method here to determine if the + // incomming message is at the 'front' or not. } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntry.java b/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntry.java index 09600b9d28..25d41c8203 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntry.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntry.java @@ -119,16 +119,42 @@ public interface QueueEntry extends Comparable<QueueEntry>, Filterable<AMQExcept final static EntryState EXPIRED_STATE = new ExpiredState(); final static EntryState NON_SUBSCRIPTION_ACQUIRED_STATE = new NonSubscriptionAcquiredState(); + /** Flag to indicate that this message requires 'immediate' delivery. */ + + final static byte IMMEDIATE = 0x01; + + /** + * Flag to indicate whether this message has been delivered to a consumer. Used in implementing return functionality + * for messages published with the 'immediate' flag. + */ + + final static byte DELIVERED_TO_CONSUMER = 0x02; + + AMQQueue getQueue(); AMQMessage getMessage(); long getSize(); + /** + * Called selectors to determin if the message has already been sent + * + * @return _deliveredToConsumer + */ boolean getDeliveredToConsumer(); + /** + * Checks to see if the message has expired. If it has the message is dequeued. + * + * @return true if the message has expire + * + * @throws org.apache.qpid.AMQException + */ boolean expired() throws AMQException; + public void setExpiration(final long expiration); + boolean isAcquired(); boolean acquire(); @@ -143,10 +169,22 @@ public interface QueueEntry extends Comparable<QueueEntry>, Filterable<AMQExcept void setDeliveredToSubscription(); + /** + * Called when this message is delivered to a consumer. (used to implement the 'immediate' flag functionality). + * And for selector efficiency. + */ + public void setDeliveredToConsumer(); + void release(); String debugIdentity(); + /** + * Called to enforce the 'immediate' flag. + * + * @returns true if the message is marked for immediate delivery but has not been marked as delivered + * to a consumer + */ boolean immediateAndNotDelivered(); void setRedelivered(boolean b); @@ -180,4 +218,11 @@ public interface QueueEntry extends Comparable<QueueEntry>, Filterable<AMQExcept void addStateChangeListener(StateChangeListener listener); boolean removeStateChangeListener(StateChangeListener listener); -} + + void flow() throws UnableToFlowMessageException; + + void recover(); + + boolean isFlowed(); + +}
\ No newline at end of file diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java b/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java index 911ed8321b..3d464d01d3 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java @@ -30,6 +30,7 @@ import java.util.Set; import java.util.HashSet; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import java.util.concurrent.atomic.AtomicLongFieldUpdater; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.CopyOnWriteArraySet; @@ -78,6 +79,17 @@ public class QueueEntryImpl implements QueueEntry volatile QueueEntryImpl _next; + private long _messageSize; + private QueueBackingStore _backingStore; + private AtomicBoolean _flowed; + private Long _messageId; + + private byte _flags = 0; + + private long _expiration; + + private static final byte IMMEDIATE_AND_DELIVERED = (byte) (IMMEDIATE | DELIVERED_TO_CONSUMER); + QueueEntryImpl(SimpleQueueEntryList queueEntryList) { @@ -88,8 +100,7 @@ public class QueueEntryImpl implements QueueEntry public QueueEntryImpl(SimpleQueueEntryList queueEntryList, AMQMessage message, final long entryId) { - _queueEntryList = queueEntryList; - _message = message; + this(queueEntryList,message); _entryIdUpdater.set(this, entryId); } @@ -98,6 +109,19 @@ public class QueueEntryImpl implements QueueEntry { _queueEntryList = queueEntryList; _message = message; + if (message != null) + { + _messageId = message.getMessageId(); + _messageSize = message.getSize(); + + if(message.isImmediate()) + { + _flags |= IMMEDIATE; + } + _expiration = message.getExpiration(); + } + _backingStore = queueEntryList.getBackingStore(); + _flowed = new AtomicBoolean(false); } protected void setEntryId(long entryId) @@ -122,17 +146,34 @@ public class QueueEntryImpl implements QueueEntry public long getSize() { - return getMessage().getSize(); + return _messageSize; } public boolean getDeliveredToConsumer() { - return getMessage().getDeliveredToConsumer(); + return (_flags & DELIVERED_TO_CONSUMER) != 0; + } + + public void setDeliveredToConsumer() + { + _flags |= DELIVERED_TO_CONSUMER; } public boolean expired() throws AMQException { - return getMessage().expired(); + if (_expiration != 0L) + { + long now = System.currentTimeMillis(); + + return (now > _expiration); + } + + return false; + } + + public void setExpiration(final long expiration) + { + _expiration = expiration; } public boolean isAcquired() @@ -169,7 +210,7 @@ public class QueueEntryImpl implements QueueEntry public void setDeliveredToSubscription() { - getMessage().setDeliveredToConsumer(); + _flags |= DELIVERED_TO_CONSUMER; } public void release() @@ -185,7 +226,7 @@ public class QueueEntryImpl implements QueueEntry public boolean immediateAndNotDelivered() { - return _message.immediateAndNotDelivered(); + return (_flags & IMMEDIATE_AND_DELIVERED) == IMMEDIATE; } public ContentHeaderBody getContentHeaderBody() throws AMQException @@ -206,8 +247,8 @@ public class QueueEntryImpl implements QueueEntry public void setRedelivered(boolean redelivered) { _redelivered = redelivered; - // todo - here we could mark this message as redelivered so we don't have to mark - // all messages on recover as redelivered. + // todo - here we could record this message as redelivered on this queue in the transactionLog + // so we don't have to mark all messages on recover as redelivered. } public Subscription getDeliveredSubscription() @@ -281,6 +322,8 @@ public class QueueEntryImpl implements QueueEntry s.restoreCredit(this); } + _queueEntryList.dequeued(this); + getQueue().dequeue(storeContext, this); if (_stateChangeListeners != null) @@ -337,6 +380,34 @@ public class QueueEntryImpl implements QueueEntry return false; } + public void flow() throws UnableToFlowMessageException + { + if (_message != null && _backingStore != null) + { + if(_log.isDebugEnabled()) + { + _log.debug("Flowing message:" + _message.debugIdentity()); + } + _backingStore.flow(_message); + _message = null; + _flowed.getAndSet(true); + } + } + + public void recover() + { + if (_messageId != null && _backingStore != null) + { + _message = _backingStore.recover(_messageId); + _flowed.getAndSet(false); + } + } + + public boolean isFlowed() + { + return _flowed.get(); + } + public int compareTo(final QueueEntry o) { @@ -382,7 +453,11 @@ public class QueueEntryImpl implements QueueEntry if(state != DELETED_STATE && _stateUpdater.compareAndSet(this,state,DELETED_STATE)) { - _queueEntryList.advanceHead(); + _queueEntryList.advanceHead(); + if (_backingStore != null) + { + _backingStore.delete(_messageId); + } return true; } else @@ -395,4 +470,6 @@ public class QueueEntryImpl implements QueueEntry { return _queueEntryList; } + + } diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryList.java b/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryList.java index 313e076f61..72783e3f78 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryList.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryList.java @@ -20,7 +20,7 @@ */ package org.apache.qpid.server.queue; -public interface QueueEntryList +public interface QueueEntryList extends FlowableQueueEntryList { AMQQueue getQueue(); diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java index a4945bc11a..7f46a6063a 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java @@ -72,12 +72,6 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener private final List<Task> _deleteTaskList = new CopyOnWriteArrayList<Task>(); - private final AtomicInteger _atomicQueueCount = new AtomicInteger(0); - - private final AtomicLong _atomicQueueSize = new AtomicLong(0L); - - private final AtomicLong _atomicQueueInMemory = new AtomicLong(0L); - private final AtomicInteger _activeSubscriberCount = new AtomicInteger(); protected final SubscriptionList _subscriptionList = new SubscriptionList(this); @@ -106,11 +100,6 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener /** the minimum interval between sending out consecutive alerts of the same type */ public long _minimumAlertRepeatGap = ApplicationRegistry.getInstance().getConfiguration().getMinimumAlertRepeatGap(); - /** The maximum amount of memory that is allocated to this queue. Beyond this the queue will flow to disk. */ - private long _memoryUsageMaximum = 0; - - /** The minimum amount of memory that is allocated to this queue. If the queueDepth hits this level then more flowed data can be read in. */ - private long _memoryUsageMinimum = 0; private static final int MAX_ASYNC_DELIVERIES = 10; @@ -120,8 +109,6 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener private AtomicReference _asynchronousRunner = new AtomicReference(null); private AtomicInteger _deliveredMessages = new AtomicInteger(); private AtomicBoolean _stopped = new AtomicBoolean(false); - /** Control to determin if this queue is flowed or not. */ - protected AtomicBoolean _flowed = new AtomicBoolean(false); protected SimpleAMQQueue(AMQShortString name, boolean durable, AMQShortString owner, boolean autoDelete, VirtualHost virtualHost) throws AMQException @@ -168,13 +155,6 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener } resetNotifications(); - resetFlowToDisk(); - } - - public void resetFlowToDisk() - { - setMemoryUsageMaximum(_memoryUsageMaximum); - setMemoryUsageMinimum(_memoryUsageMinimum); } public void resetNotifications() @@ -205,7 +185,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener public boolean isFlowed() { - return _flowed.get(); + return _entries.isFlowed(); } public AMQShortString getOwner() @@ -341,10 +321,6 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener public QueueEntry enqueue(StoreContext storeContext, AMQMessage message) throws AMQException { - - incrementQueueCount(); - incrementQueueSize(message); - _totalMessagesReceived.incrementAndGet(); QueueEntry entry; @@ -485,17 +461,6 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener // Simple Queues don't :-) } - private void incrementQueueSize(final AMQMessage message) - { - getAtomicQueueSize().addAndGet(message.getSize()); - getAtomicQueueInMemory().addAndGet(message.getSize()); - } - - private void incrementQueueCount() - { - getAtomicQueueCount().incrementAndGet(); - } - private void deliverMessage(final Subscription sub, final QueueEntry entry) throws AMQException { @@ -594,8 +559,6 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener */ public void dequeue(StoreContext storeContext, QueueEntry entry) throws FailedDequeueException { - decrementQueueCount(); - decrementQueueSize(entry); if (entry.acquiredBySubscription()) { _deliveredMessages.decrementAndGet(); @@ -625,16 +588,6 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener } - private void decrementQueueSize(final QueueEntry entry) - { - getAtomicQueueSize().addAndGet(-entry.getMessage().getSize()); - getAtomicQueueInMemory().addAndGet(-entry.getMessage().getSize()); - } - - void decrementQueueCount() - { - getAtomicQueueCount().decrementAndGet(); - } public boolean resend(final QueueEntry entry, final Subscription subscription) throws AMQException { @@ -682,17 +635,17 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener public long getMemoryUsageCurrent() { - return getAtomicQueueInMemory().get(); + return getQueueInMemory(); } public int getMessageCount() { - return getAtomicQueueCount().get(); + return getQueueCount(); } public long getQueueDepth() { - return getAtomicQueueSize().get(); + return getQueueSize(); } public int getUndeliveredMessageCount() @@ -768,21 +721,20 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener return _name.compareTo(o.getName()); } - public AtomicInteger getAtomicQueueCount() + public int getQueueCount() { - return _atomicQueueCount; + return _entries.size(); } - public AtomicLong getAtomicQueueSize() + public long getQueueSize() { - return _atomicQueueSize; + return _entries.dataSize(); } - public AtomicLong getAtomicQueueInMemory() + public long getQueueInMemory() { - return _atomicQueueInMemory; - } - + return _entries.memoryUsed(); + } private boolean isExclusiveSubscriber() { @@ -1493,46 +1445,22 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener public long getMemoryUsageMaximum() { - return _memoryUsageMaximum; + return _entries.getMemoryUsageMaximum(); } public void setMemoryUsageMaximum(long maximumMemoryUsage) { - _memoryUsageMaximum = maximumMemoryUsage; - - // Don't attempt to start the inhaler/purger unless we have a minimum value specified. - if (_memoryUsageMaximum > 0) - { - // If we've increased the max memory above what we have in memory then we can inhale more - if (_memoryUsageMaximum > _atomicQueueInMemory.get()) - { - //TODO start inhaler - } - else // if we have now have to much memory in use we need to purge. - { - //TODO start purger - } - } + _entries.setMemoryUsageMaximum(maximumMemoryUsage); } public long getMemoryUsageMinimum() { - return _memoryUsageMinimum; + return _entries.getMemoryUsageMinimum(); } public void setMemoryUsageMinimum(long minimumMemoryUsage) { - _memoryUsageMinimum = minimumMemoryUsage; - - // Don't attempt to start the inhaler unless we have a minimum value specified. - if (_memoryUsageMinimum > 0) - { - // If we've increased the minimum memory above what we have in memory then we need to inhale more - if (_memoryUsageMinimum >= _atomicQueueInMemory.get()) - { - //TODO start inhaler - } - } + _entries.setMemoryUsageMinimum(minimumMemoryUsage); } public long getMinimumAlertRepeatGap() diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java b/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java index a46c5ae2e8..10abdd8318 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java @@ -1,6 +1,9 @@ package org.apache.qpid.server.queue; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; /* * @@ -22,8 +25,9 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; * under the License. * */ -public class SimpleQueueEntryList implements QueueEntryList +public class SimpleQueueEntryList extends FlowableBaseQueueEntryList implements QueueEntryList { + private final QueueEntryImpl _head; private volatile QueueEntryImpl _tail; @@ -41,12 +45,9 @@ public class SimpleQueueEntryList implements QueueEntryList AtomicReferenceFieldUpdater.newUpdater (QueueEntryImpl.class, QueueEntryImpl.class, "_next"); - - - - public SimpleQueueEntryList(AMQQueue queue) { + super(queue); _queue = queue; _head = new QueueEntryImpl(this); _tail = _head; @@ -77,6 +78,9 @@ public class SimpleQueueEntryList implements QueueEntryList public QueueEntry add(AMQMessage message) { QueueEntryImpl node = new QueueEntryImpl(this, message); + + incrementCounters(node); + for (;;) { QueueEntryImpl tail = _tail; @@ -101,12 +105,12 @@ public class SimpleQueueEntryList implements QueueEntryList } } + public QueueEntry next(QueueEntry node) { return ((QueueEntryImpl)node).getNext(); } - public class QueueEntryIteratorImpl implements QueueEntryIterator { @@ -172,7 +176,9 @@ public class SimpleQueueEntryList implements QueueEntryList { return new SimpleQueueEntryList(queue); } + } - + + } diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/TransientAMQMessage.java b/java/broker/src/main/java/org/apache/qpid/server/queue/TransientAMQMessage.java index 0334a54fab..4c9fe81439 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/TransientAMQMessage.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/TransientAMQMessage.java @@ -54,23 +54,11 @@ public class TransientAMQMessage implements AMQMessage protected final Long _messageId; - /** Flag to indicate that this message requires 'immediate' delivery. */ - - private static final byte IMMEDIATE = 0x01; - - /** - * Flag to indicate whether this message has been delivered to a consumer. Used in implementing return functionality - * for messages published with the 'immediate' flag. - */ - - private static final byte DELIVERED_TO_CONSUMER = 0x02; private byte _flags = 0; - private long _expiration; - private AMQProtocolSession.ProtocolSessionIdentifier _sessionIdentifier; - private static final byte IMMEDIATE_AND_DELIVERED = (byte) (IMMEDIATE | DELIVERED_TO_CONSUMER); + private long _expiration; /** * Used to iterate through all the body frames associated with this message. Will not keep all the data in memory @@ -165,11 +153,16 @@ public class TransientAMQMessage implements AMQMessage return "(HC:" + System.identityHashCode(this) + " ID:" + getMessageId() +")"; } - public void setExpiration(final long expiration) + public void setExpiration(long expiration) { _expiration = expiration; } + public long getExpiration() + { + return _expiration; + } + public Iterator<AMQDataBlock> getBodyFrameIterator(AMQProtocolSession protocolSession, int channel) { return new BodyFrameIterator(protocolSession, channel); @@ -190,57 +183,6 @@ public class TransientAMQMessage implements AMQMessage return _messageId; } - /** - * Called selectors to determin if the message has already been sent - * - * @return _deliveredToConsumer - */ - public boolean getDeliveredToConsumer() - { - return (_flags & DELIVERED_TO_CONSUMER) != 0; - } - - /** - * Called to enforce the 'immediate' flag. - * - * @returns true if the message is marked for immediate delivery but has not been marked as delivered - * to a consumer - */ - public boolean immediateAndNotDelivered() - { - - return (_flags & IMMEDIATE_AND_DELIVERED) == IMMEDIATE; - - } - - /** - * Checks to see if the message has expired. If it has the message is dequeued. - * - * @return true if the message has expire - * - * @throws AMQException - */ - public boolean expired() throws AMQException - { - - if (_expiration != 0L) - { - long now = System.currentTimeMillis(); - - return (now > _expiration); - } - - return false; - } - - /** - * Called when this message is delivered to a consumer. (used to implement the 'immediate' flag functionality). - * And for selector efficiency. - */ - public void setDeliveredToConsumer() - { - _flags |= DELIVERED_TO_CONSUMER; - } public long getSize() { @@ -315,6 +257,11 @@ public class TransientAMQMessage implements AMQMessage return false; } + public boolean isImmediate() + { + return _messagePublishInfo.isImmediate(); + } + /** * This is called when all the content has been received. * @@ -366,11 +313,6 @@ public class TransientAMQMessage implements AMQMessage { _contentBodies = Collections.EMPTY_LIST; } - - if (_messagePublishInfo.isImmediate()) - { - _flags |= IMMEDIATE; - } } public void recoverContentBodyFrame(ContentChunk contentChunk, boolean isLastContentBody) throws AMQException diff --git a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java index b4b392c91d..c5b6eeca3e 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java +++ b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java @@ -48,6 +48,8 @@ import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.queue.AMQQueueFactory; import org.apache.qpid.server.queue.DefaultQueueRegistry; import org.apache.qpid.server.queue.QueueRegistry; +import org.apache.qpid.server.queue.QueueBackingStore; +import org.apache.qpid.server.queue.FileQueueBackingStore; import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.routing.RoutingTable; import org.apache.qpid.server.security.access.ACLManager; @@ -86,6 +88,7 @@ public class VirtualHost implements Accessable private final Timer _houseKeepingTimer; private VirtualHostConfiguration _configuration; + private QueueBackingStore _queueBackingStore; public void setAccessableName(String name) { @@ -113,6 +116,11 @@ public class VirtualHost implements Accessable return _configuration ; } + public QueueBackingStore getQueueBackingStore() + { + return _queueBackingStore; + } + /** * Abstract MBean class. This has some of the methods implemented from management intrerface for exchanges. Any * implementaion of an Exchange MBean should extend this class. @@ -186,6 +194,9 @@ public class VirtualHost implements Accessable initialiseRoutingTable(hostConfig); } + _queueBackingStore = new FileQueueBackingStore(); + _queueBackingStore.configure(this,hostConfig); + _exchangeFactory.initialise(hostConfig); _exchangeRegistry.initialise(); diff --git a/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java b/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java index d46ba85069..49afcb1340 100644 --- a/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java +++ b/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java @@ -352,7 +352,7 @@ public class Show extends AbstractCommand isredelivered.add(entry.isRedelivered() ? "true" : "false"); - isdelivered.add(msg.getDeliveredToConsumer() ? "true" : "false"); + isdelivered.add(entry.getDeliveredToConsumer() ? "true" : "false"); BasicContentHeaderProperties headers = null; diff --git a/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java b/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java index 78cf610f28..6021f100f5 100644 --- a/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java +++ b/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java @@ -40,6 +40,7 @@ import org.apache.qpid.server.queue.MessageCleanupException; import org.apache.qpid.server.queue.MockProtocolSession; import org.apache.qpid.server.queue.QueueEntry; import org.apache.qpid.server.queue.SimpleAMQQueue; +import org.apache.qpid.server.queue.UnableToFlowMessageException; import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.store.MemoryMessageStore; import org.apache.qpid.server.store.SkeletonMessageStore; @@ -229,6 +230,11 @@ public class AbstractHeadersExchangeTestBase extends TestCase return false; //To change body of implemented methods use File | Settings | File Templates. } + public void setExpiration(long expiration) + { + //To change body of implemented methods use File | Settings | File Templates. + } + public boolean isAcquired() { return false; //To change body of implemented methods use File | Settings | File Templates. @@ -264,6 +270,11 @@ public class AbstractHeadersExchangeTestBase extends TestCase //To change body of implemented methods use File | Settings | File Templates. } + public void setDeliveredToConsumer() + { + //To change body of implemented methods use File | Settings | File Templates. + } + public void release() { //To change body of implemented methods use File | Settings | File Templates. @@ -314,32 +325,38 @@ public class AbstractHeadersExchangeTestBase extends TestCase //To change body of implemented methods use File | Settings | File Templates. } - public void dispose(final StoreContext storeContext) throws MessageCleanupException + + public void dequeueAndDelete(StoreContext storeContext) throws FailedDequeueException { //To change body of implemented methods use File | Settings | File Templates. } - public void restoreCredit() + public boolean isQueueDeleted() { - //To change body of implemented methods use File | Settings | File Templates. + return false; //To change body of implemented methods use File | Settings | File Templates. } - public void dequeueAndDelete(StoreContext storeContext) throws FailedDequeueException + public void addStateChangeListener(StateChangeListener listener) { //To change body of implemented methods use File | Settings | File Templates. } - public boolean isQueueDeleted() + public boolean removeStateChangeListener(StateChangeListener listener) { return false; //To change body of implemented methods use File | Settings | File Templates. } - public void addStateChangeListener(StateChangeListener listener) + public void flow() throws UnableToFlowMessageException { //To change body of implemented methods use File | Settings | File Templates. } - public boolean removeStateChangeListener(StateChangeListener listener) + public void recover() + { + //To change body of implemented methods use File | Settings | File Templates. + } + + public boolean isFlowed() { return false; //To change body of implemented methods use File | Settings | File Templates. } diff --git a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java index ba02e6f6bd..a11e60d7de 100644 --- a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java +++ b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java @@ -102,4 +102,11 @@ public class AMQPriorityQueueTest extends SimpleAMQQueueTest return message; } + + @Override + public void testMessagesFlowToDisk() throws AMQException, InterruptedException + { + //Disable this test pending completion of QPID-1637 + } + } diff --git a/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java b/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java index cc6c486e11..b38da53406 100644 --- a/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java +++ b/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java @@ -22,6 +22,7 @@ package org.apache.qpid.server.queue; import org.apache.qpid.server.store.StoreContext; import org.apache.qpid.AMQException; +import org.apache.qpid.framing.abstraction.MessagePublishInfoImpl; public class MockAMQMessage extends TransientAMQMessage { @@ -29,6 +30,7 @@ public class MockAMQMessage extends TransientAMQMessage throws AMQException { super(messageId); + _messagePublishInfo = new MessagePublishInfoImpl(null,false,false,null); } diff --git a/java/broker/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTest.java b/java/broker/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTest.java index f7cd860c22..9e12e1bef7 100644 --- a/java/broker/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTest.java +++ b/java/broker/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTest.java @@ -21,16 +21,25 @@ package org.apache.qpid.server.queue; import junit.framework.TestCase; +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.BasicContentHeaderProperties; +import org.apache.qpid.framing.ContentHeaderBody; +import org.apache.qpid.framing.ContentHeaderProperties; +import org.apache.qpid.framing.abstraction.ContentChunk; +import org.apache.qpid.framing.abstraction.MessagePublishInfo; +import org.apache.qpid.framing.abstraction.MessagePublishInfoImpl; +import org.apache.qpid.server.store.StoreContext; + +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; public class QueueEntryImplTest extends TestCase { - /** - * Test the Redelivered state of a QueueEntryImpl - */ + /** Test the Redelivered state of a QueueEntryImpl */ public void testRedelivered() { - QueueEntry entry = new QueueEntryImpl(null, null); + QueueEntry entry = new MockQueueEntry(null); assertFalse("New message should not be redelivered", entry.isRedelivered()); @@ -45,5 +54,187 @@ public class QueueEntryImplTest extends TestCase } + public void testImmediateAndNotDelivered() + { + AMQMessage message = MessageFactory.getInstance().createMessage(null, false); + + MessagePublishInfo mpi = new MessagePublishInfoImpl(null, true, false, null); + int bodySize = 0; + + BasicContentHeaderProperties props = new BasicContentHeaderProperties(); + + props.setAppId("HandleTest"); + + ContentHeaderBody chb = new ContentHeaderBody(0, 0, props, bodySize); + + try + { + message.setPublishAndContentHeaderBody(null, mpi, chb); + + QueueEntry queueEntry = new MockQueueEntry(message); + + assertTrue("Undelivered Immediate message should still be marked as so", queueEntry.immediateAndNotDelivered()); + + assertFalse("Undelivered Message should not say it is delivered.", queueEntry.getDeliveredToConsumer()); + + queueEntry.setDeliveredToConsumer(); + + assertTrue("Delivered Message should say it is delivered.", queueEntry.getDeliveredToConsumer()); + + assertFalse("Delivered Immediate message now be marked as so", queueEntry.immediateAndNotDelivered()); + } + catch (AMQException e) + { + fail(e.getMessage()); + } + } + + public void testNotImmediateAndNotDelivered() + { + AMQMessage message = MessageFactory.getInstance().createMessage(null, false); + + MessagePublishInfo mpi = new MessagePublishInfoImpl(null, false, false, null); + int bodySize = 0; + + BasicContentHeaderProperties props = new BasicContentHeaderProperties(); + + props.setAppId("HandleTest"); + + ContentHeaderBody chb = new ContentHeaderBody(0, 0, props, bodySize); + + try + { + message.setPublishAndContentHeaderBody(null, mpi, chb); + + QueueEntry queueEntry = new MockQueueEntry(message); + + assertFalse("Undelivered Non-Immediate message should not result in true.", queueEntry.immediateAndNotDelivered()); + + assertFalse("Undelivered Message should not say it is delivered.", queueEntry.getDeliveredToConsumer()); + + queueEntry.setDeliveredToConsumer(); + + assertTrue("Delivered Message should say it is delivered.", queueEntry.getDeliveredToConsumer()); + + assertFalse("Delivered Non-Immediate message not change this return", queueEntry.immediateAndNotDelivered()); + } + catch (AMQException e) + { + fail(e.getMessage()); + } + } + + public void testExpiry() + { + AMQMessage message = MessageFactory.getInstance().createMessage(null, false); + + MessagePublishInfo mpi = new MessagePublishInfoImpl(null, false, false, null); + int bodySize = 0; + + BasicContentHeaderProperties props = new BasicContentHeaderProperties(); + + props.setAppId("HandleTest"); + + ContentHeaderBody chb = new ContentHeaderBody(0, 0, props, bodySize); + + ReentrantLock waitLock = new ReentrantLock(); + Condition wait = waitLock.newCondition(); + try + { + message.setExpiration(System.currentTimeMillis() + 10L); + + message.setPublishAndContentHeaderBody(null, mpi, chb); + + QueueEntry queueEntry = new MockQueueEntry(message); + + assertFalse("New messages should not be expired.", queueEntry.expired()); + + final long MILLIS = 1000000L; + long waitTime = 20 * MILLIS; + + while (waitTime > 0) + { + try + { + waitLock.lock(); + + waitTime = wait.awaitNanos(waitTime); + } + catch (InterruptedException e) + { + //Stop if we are interrupted + fail(e.getMessage()); + } + finally + { + waitLock.unlock(); + } + + } + + assertTrue("After a sleep messages should now be expired.", queueEntry.expired()); + + } + catch (AMQException e) + { + fail(e.getMessage()); + } + } + + public void testNoExpiry() + { + AMQMessage message = MessageFactory.getInstance().createMessage(null, false); + + MessagePublishInfo mpi = new MessagePublishInfoImpl(null, false, false, null); + int bodySize = 0; + + BasicContentHeaderProperties props = new BasicContentHeaderProperties(); + + props.setAppId("HandleTest"); + + ContentHeaderBody chb = new ContentHeaderBody(0, 0, props, bodySize); + + ReentrantLock waitLock = new ReentrantLock(); + Condition wait = waitLock.newCondition(); + try + { + + message.setPublishAndContentHeaderBody(null, mpi, chb); + + QueueEntry queueEntry = new MockQueueEntry(message); + + assertFalse("New messages should not be expired.", queueEntry.expired()); + + final long MILLIS = 1000000L; + long waitTime = 10 * MILLIS; + + while (waitTime > 0) + { + try + { + waitLock.lock(); + + waitTime = wait.awaitNanos(waitTime); + } + catch (InterruptedException e) + { + //Stop if we are interrupted + fail(e.getMessage()); + } + finally + { + waitLock.unlock(); + } + + } + + assertFalse("After a sleep messages without an expiry should not expire.", queueEntry.expired()); + + } + catch (AMQException e) + { + fail(e.getMessage()); + } + } } diff --git a/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java b/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java index 7a97837208..9a5f7f20c6 100644 --- a/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java +++ b/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java @@ -21,8 +21,6 @@ package org.apache.qpid.server.queue; */ import junit.framework.TestCase; - -import org.apache.commons.configuration.Configuration; import org.apache.commons.configuration.PropertiesConfiguration; import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQShortString; @@ -35,13 +33,13 @@ import org.apache.qpid.server.configuration.VirtualHostConfiguration; import org.apache.qpid.server.exchange.DirectExchange; import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.store.StoreContext; -import org.apache.qpid.server.store.TestableMemoryMessageStore; import org.apache.qpid.server.store.TestTransactionLog; +import org.apache.qpid.server.store.TestableMemoryMessageStore; import org.apache.qpid.server.subscription.MockSubscription; import org.apache.qpid.server.subscription.Subscription; import org.apache.qpid.server.txn.NonTransactionalContext; +import org.apache.qpid.server.txn.TransactionalContext; import org.apache.qpid.server.virtualhost.VirtualHost; -import org.apache.qpid.server.transactionlog.TransactionLog; import java.util.ArrayList; import java.util.List; @@ -51,7 +49,7 @@ public class SimpleAMQQueueTest extends TestCase protected SimpleAMQQueue _queue; protected VirtualHost _virtualHost; - protected TestableMemoryMessageStore _store = new TestableMemoryMessageStore(); + protected TestableMemoryMessageStore _transactionLog = new TestableMemoryMessageStore(); protected AMQShortString _qname = new AMQShortString("qname"); protected AMQShortString _owner = new AMQShortString("owner"); protected AMQShortString _routingKey = new AMQShortString("routing key"); @@ -70,7 +68,7 @@ public class SimpleAMQQueueTest extends TestCase ApplicationRegistry applicationRegistry = (ApplicationRegistry) ApplicationRegistry.getInstance(1); PropertiesConfiguration env = new PropertiesConfiguration(); - _virtualHost = new VirtualHost(new VirtualHostConfiguration(getClass().getName(), env), _store); + _virtualHost = new VirtualHost(new VirtualHostConfiguration(getClass().getName(), env), _transactionLog); applicationRegistry.getVirtualHostRegistry().registerVirtualHost(_virtualHost); _queue = (SimpleAMQQueue) AMQQueueFactory.createAMQQueueImpl(_qname, false, _owner, false, _virtualHost, _arguments); @@ -320,8 +318,8 @@ public class SimpleAMQQueueTest extends TestCase public void testEnqueueDequeueOfPersistentMessageToNonDurableQueue() throws AMQException { // Create IncomingMessage and nondurable queue - NonTransactionalContext txnContext = new NonTransactionalContext(_store, null, null, null); - IncomingMessage msg = new IncomingMessage(info, txnContext, new MockProtocolSession(_store), _store); + NonTransactionalContext txnContext = new NonTransactionalContext(_transactionLog, null, null, null); + IncomingMessage msg = new IncomingMessage(info, txnContext, new MockProtocolSession(_transactionLog), _transactionLog); ContentHeaderBody contentHeaderBody = new ContentHeaderBody(); contentHeaderBody.properties = new BasicContentHeaderProperties(); @@ -335,18 +333,18 @@ public class SimpleAMQQueueTest extends TestCase // Send persistent message qs.add(_queue); msg.enqueue(qs); - msg.routingComplete(_store); + msg.routingComplete(_transactionLog); - _store.storeMessageMetaData(null, messageId, new MessageMetaData(info, contentHeaderBody, 1)); + _transactionLog.storeMessageMetaData(null, messageId, new MessageMetaData(info, contentHeaderBody, 1)); // Check that it is enqueued - List<AMQQueue> data = _store.getMessageReferenceMap(messageId); + List<AMQQueue> data = _transactionLog.getMessageReferenceMap(messageId); assertNotNull(data); // Dequeue message ContentHeaderBody header = new ContentHeaderBody(); header.bodySize = MESSAGE_SIZE; - AMQMessage message = new MockPersistentAMQMessage(msg.getMessageId(), _store); + AMQMessage message = new MockPersistentAMQMessage(msg.getMessageId(), _transactionLog); message.setPublishAndContentHeaderBody(new StoreContext(), info, header); MockQueueEntry entry = new MockQueueEntry(message, _queue); @@ -355,10 +353,97 @@ public class SimpleAMQQueueTest extends TestCase entry.dequeue(null); // Check that it is dequeued - data = _store.getMessageReferenceMap(messageId); + data = _transactionLog.getMessageReferenceMap(messageId); assertNull(data); } + public void testMessagesFlowToDisk() throws AMQException, InterruptedException + { + // Create IncomingMessage and nondurable queue + NonTransactionalContext txnContext = new NonTransactionalContext(_transactionLog, null, null, null); + + //Set the Memory Usage to be very low + _queue.setMemoryUsageMaximum(10); + + for (int msgCount = 0; msgCount < 10; msgCount++) + { + sendMessage(txnContext); + } + + //Check that we can hold 10 messages without flowing + assertEquals(10, _queue.getMessageCount()); + assertEquals(10, _queue.getMemoryUsageCurrent()); + assertTrue("Queue is flowed.", !_queue.isFlowed()); + + // Send anothe and ensure we are flowed + sendMessage(txnContext); + assertEquals(11, _queue.getMessageCount()); + assertEquals(10, _queue.getMemoryUsageCurrent()); + assertTrue("Queue is not flowed.", _queue.isFlowed()); + + //send another 9 so there are 20msgs in total on the queue + for (int msgCount = 0; msgCount < 9; msgCount++) + { + sendMessage(txnContext); + } + assertEquals(20, _queue.getMessageCount()); + assertEquals(10, _queue.getMemoryUsageCurrent()); + assertTrue("Queue is not flowed.", _queue.isFlowed()); + + _queue.registerSubscription(_subscription, false); + + Thread.sleep(200); + + //Ensure the messages are retreived + assertEquals("Not all messages were received.", 20, _subscription.getMessages().size()); + + //Ensure we got the content + for (int index = 0; index < 10; index++) + { + QueueEntry entry = _subscription.getMessages().get(index); + assertNotNull("Message:" + index + " was null.", entry.getMessage()); + assertTrue(!entry.isFlowed()); + } + + //ensure we were received 10 flowed messages + for (int index = 10; index < 20; index++) + { + QueueEntry entry = _subscription.getMessages().get(index); + assertNull("Message:" + index + " was not null.", entry.getMessage()); + assertTrue(entry.isFlowed()); + } + } + + private void sendMessage(TransactionalContext txnContext) throws AMQException + { + IncomingMessage msg = new IncomingMessage(info, txnContext, new MockProtocolSession(_transactionLog), _transactionLog); + + ContentHeaderBody contentHeaderBody = new ContentHeaderBody(); + contentHeaderBody.bodySize = 1; + contentHeaderBody.properties = new BasicContentHeaderProperties(); + ((BasicContentHeaderProperties) contentHeaderBody.properties).setDeliveryMode((byte) 2); + msg.setContentHeaderBody(contentHeaderBody); + + long messageId = msg.getMessageId(); + + ArrayList<AMQQueue> qs = new ArrayList<AMQQueue>(); + + // Send persistent 10 messages + + qs.add(_queue); + msg.enqueue(qs); + + msg.routingComplete(_transactionLog); + + msg.addContentBodyFrame(new MockContentChunk(1)); + + msg.deliverToQueues(); + + //Check message was correctly enqueued + List<AMQQueue> data = _transactionLog.getMessageReferenceMap(messageId); + assertNotNull(data); + } + // FIXME: move this to somewhere useful private static AMQMessage createMessage(final MessagePublishInfo publishBody) { @@ -384,7 +469,7 @@ public class SimpleAMQQueueTest extends TestCase public AMQMessage createMessage() throws AMQException { - AMQMessage message = new TestMessage(info, _store); + AMQMessage message = new TestMessage(info, _transactionLog); ContentHeaderBody header = new ContentHeaderBody(); header.bodySize = MESSAGE_SIZE; @@ -410,7 +495,6 @@ public class SimpleAMQQueueTest extends TestCase _transactionLog = transactionLog; } - void assertCountEquals(int expected) { assertEquals("Wrong count for message with tag " + _tag, expected, diff --git a/java/broker/src/test/java/org/apache/qpid/server/queue/TransientMessageTest.java b/java/broker/src/test/java/org/apache/qpid/server/queue/TransientMessageTest.java index 16d1ab60f3..6fd153f398 100644 --- a/java/broker/src/test/java/org/apache/qpid/server/queue/TransientMessageTest.java +++ b/java/broker/src/test/java/org/apache/qpid/server/queue/TransientMessageTest.java @@ -287,180 +287,5 @@ public class TransientMessageTest extends TestCase assertFalse(_message.isPersistent()); } - public void testImmediateAndNotDelivered() - { - _message = newMessage(); - - MessagePublishInfo mpi = new MessagePublishInfoImpl(null, true, false, null); - int bodySize = 0; - - BasicContentHeaderProperties props = new BasicContentHeaderProperties(); - - props.setAppId("HandleTest"); - - ContentHeaderBody chb = new ContentHeaderBody(0, 0, props, bodySize); - - try - { - _message.setPublishAndContentHeaderBody(_storeContext, mpi, chb); - - assertTrue("Undelivered Immediate message should still be marked as so", _message.immediateAndNotDelivered()); - - assertFalse("Undelivered Message should not say it is delivered.", _message.getDeliveredToConsumer()); - - _message.setDeliveredToConsumer(); - - assertTrue("Delivered Message should say it is delivered.", _message.getDeliveredToConsumer()); - - assertFalse("Delivered Immediate message now be marked as so", _message.immediateAndNotDelivered()); - } - catch (AMQException e) - { - fail(e.getMessage()); - } - } - - public void testNotImmediateAndNotDelivered() - { - _message = newMessage(); - - MessagePublishInfo mpi = new MessagePublishInfoImpl(null, false, false, null); - int bodySize = 0; - - BasicContentHeaderProperties props = new BasicContentHeaderProperties(); - - props.setAppId("HandleTest"); - - ContentHeaderBody chb = new ContentHeaderBody(0, 0, props, bodySize); - - try - { - _message.setPublishAndContentHeaderBody(_storeContext, mpi, chb); - - assertFalse("Undelivered Non-Immediate message should not result in true.", _message.immediateAndNotDelivered()); - - assertFalse("Undelivered Message should not say it is delivered.", _message.getDeliveredToConsumer()); - - _message.setDeliveredToConsumer(); - - assertTrue("Delivered Message should say it is delivered.", _message.getDeliveredToConsumer()); - - assertFalse("Delivered Non-Immediate message not change this return", _message.immediateAndNotDelivered()); - } - catch (AMQException e) - { - fail(e.getMessage()); - } - } - - public void testExpiry() - { - _message = newMessage(); - - MessagePublishInfo mpi = new MessagePublishInfoImpl(null, false, false, null); - int bodySize = 0; - - BasicContentHeaderProperties props = new BasicContentHeaderProperties(); - - props.setAppId("HandleTest"); - - ContentHeaderBody chb = new ContentHeaderBody(0, 0, props, bodySize); - - ReentrantLock waitLock = new ReentrantLock(); - Condition wait = waitLock.newCondition(); - try - { - _message.setExpiration(System.currentTimeMillis() + 10L); - - _message.setPublishAndContentHeaderBody(_storeContext, mpi, chb); - - assertFalse("New messages should not be expired.", _message.expired()); - - final long MILLIS =1000000L; - long waitTime = 20 * MILLIS; - - while (waitTime > 0) - { - try - { - waitLock.lock(); - - waitTime = wait.awaitNanos(waitTime); - } - catch (InterruptedException e) - { - //Stop if we are interrupted - fail(e.getMessage()); - } - finally - { - waitLock.unlock(); - } - - } - - assertTrue("After a sleep messages should now be expired.", _message.expired()); - - } - catch (AMQException e) - { - fail(e.getMessage()); - } - } - - - public void testNoExpiry() - { - _message = newMessage(); - - MessagePublishInfo mpi = new MessagePublishInfoImpl(null, false, false, null); - int bodySize = 0; - - BasicContentHeaderProperties props = new BasicContentHeaderProperties(); - - props.setAppId("HandleTest"); - - ContentHeaderBody chb = new ContentHeaderBody(0, 0, props, bodySize); - - ReentrantLock waitLock = new ReentrantLock(); - Condition wait = waitLock.newCondition(); - try - { - - _message.setPublishAndContentHeaderBody(_storeContext, mpi, chb); - - assertFalse("New messages should not be expired.", _message.expired()); - - final long MILLIS =1000000L; - long waitTime = 10 * MILLIS; - - while (waitTime > 0) - { - try - { - waitLock.lock(); - - waitTime = wait.awaitNanos(waitTime); - } - catch (InterruptedException e) - { - //Stop if we are interrupted - fail(e.getMessage()); - } - finally - { - waitLock.unlock(); - } - - } - - assertFalse("After a sleep messages without an expiry should not expire.", _message.expired()); - - } - catch (AMQException e) - { - fail(e.getMessage()); - } - } - + } |