You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2008/04/21 16:04:00 UTC
svn commit: r650148 [8/10] - in
/incubator/qpid/branches/broker-queue-refactor/java: broker/ broker/etc/
broker/src/main/java/org/apache/qpid/server/
broker/src/main/java/org/apache/qpid/server/ack/
broker/src/main/java/org/apache/qpid/server/configura...
Copied: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java (from r641230, incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java)
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java?p2=incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java&p1=incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java&r1=641230&r2=650148&rev=650148&view=diff
==============================================================================
--- incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java Mon Apr 21 07:03:35 2008
@@ -18,10 +18,10 @@
* under the License.
*
*/
-package org.apache.qpid.server.queue;
+package org.apache.qpid.server.subscription;
-import java.util.Queue;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
@@ -31,317 +31,306 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.output.ProtocolOutputConverter;
+import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.flow.FlowCreditManager;
import org.apache.qpid.server.filter.FilterManager;
import org.apache.qpid.server.filter.FilterManagerFactory;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.store.StoreContext;
-import org.apache.qpid.util.ConcurrentLinkedQueueAtomicSize;
-import org.apache.qpid.util.MessageQueue;
-import org.apache.qpid.util.ConcurrentLinkedMessageQueueAtomicSize;
/**
* Encapsulation of a supscription to a queue. <p/> Ties together the protocol session of a subscriber, the consumer tag
* that was given out by the broker and the channel id. <p/>
*/
-public class SubscriptionImpl implements Subscription
+public abstract class SubscriptionImpl implements Subscription, FlowCreditManager.FlowCreditManagerListener
{
+ private final AtomicBoolean _suspended = new AtomicBoolean(false);
- private static final Logger _suspensionlogger = Logger.getLogger("Suspension");
- private static final Logger _logger = Logger.getLogger(SubscriptionImpl.class);
-
- public final AMQChannel channel;
+ private StateListener _stateListener = new StateListener()
+ {
- public final AMQProtocolSession protocolSession;
+ public void stateChange(Subscription sub, State oldState, State newState)
+ {
- public final AMQShortString consumerTag;
+ }
+ };
- private final Object _sessionKey;
- private MessageQueue<QueueEntry> _messages;
+ private final AtomicReference<State> _state = new AtomicReference<State>(State.ACTIVE);
+ private final AtomicReference<Object> _queueContext = new AtomicReference<Object>(null);
- private Queue<QueueEntry> _resendQueue;
+ static final class BrowserSubscription extends SubscriptionImpl
+ {
+ public BrowserSubscription(int channelId, AMQProtocolSession protocolSession,
+ AMQShortString consumerTag, FieldTable filters,
+ boolean noLocal, FlowCreditManager creditManager)
+ throws AMQException
+ {
+ super(channelId, protocolSession, consumerTag, filters, noLocal, creditManager);
+ }
- private final boolean _noLocal;
+ public boolean isBrowser()
+ {
+ return true;
+ }
- /** True if messages need to be acknowledged */
- private final boolean _acks;
- private FilterManager _filters;
- private final boolean _isBrowser;
- private final Boolean _autoClose;
- private boolean _sentClose = false;
+ /**
+ * This method can be called by each of the publisher threads. As a result all changes to the channel object must be
+ * thread safe.
+ *
+ * @param msg The message to send
+ * @throws AMQException
+ */
+ public void send(QueueEntry msg) throws AMQException
+ {
+ // We don't decrement the reference here as we don't want to consume the message
+ // but we do want to send it to the client.
- private static final String CLIENT_PROPERTIES_INSTANCE = ClientProperties.instance.toString();
+ synchronized (getChannel())
+ {
+ long deliveryTag = getChannel().getNextDeliveryTag();
- private AMQQueue _queue;
- private final AtomicBoolean _sendLock = new AtomicBoolean(false);
+ getProtocolSession().getProtocolOutputConverter().writeDeliver(msg.getMessage(), getChannel().getChannelId(), deliveryTag, getConumerTag());
+ }
+ }
+ }
- public static class Factory implements SubscriptionFactory
+ static final class NoAckSubscription extends SubscriptionImpl
{
- public Subscription createSubscription(int channel, AMQProtocolSession protocolSession,
- AMQShortString consumerTag, boolean acks, FieldTable filters,
- boolean noLocal, AMQQueue queue) throws AMQException
+ public NoAckSubscription(int channelId, AMQProtocolSession protocolSession,
+ AMQShortString consumerTag, FieldTable filters,
+ boolean noLocal, FlowCreditManager creditManager)
+ throws AMQException
{
- return new SubscriptionImpl(channel, protocolSession, consumerTag, acks, filters, noLocal, queue);
+ super(channelId, protocolSession, consumerTag, filters, noLocal, creditManager);
}
- public SubscriptionImpl createSubscription(int channel, AMQProtocolSession protocolSession, AMQShortString consumerTag)
- throws AMQException
+
+ public boolean isBrowser()
{
- return new SubscriptionImpl(channel, protocolSession, consumerTag, false, null, false, null);
+ return false;
}
- }
-
- public SubscriptionImpl(int channelId, AMQProtocolSession protocolSession,
- AMQShortString consumerTag, boolean acks)
- throws AMQException
- {
- this(channelId, protocolSession, consumerTag, acks, null, false, null);
- }
- public SubscriptionImpl(int channelId, AMQProtocolSession protocolSession,
- AMQShortString consumerTag, boolean acks, FieldTable filters,
- boolean noLocal, AMQQueue queue)
- throws AMQException
- {
- AMQChannel channel = protocolSession.getChannel(channelId);
- if (channel == null)
+ /**
+ * This method can be called by each of the publisher threads. As a result all changes to the channel object must be
+ * thread safe.
+ *
+ * @param entry The message to send
+ * @throws AMQException
+ */
+ public void send(QueueEntry entry) throws AMQException
{
- throw new AMQException(AMQConstant.NOT_FOUND, "channel :" + channelId + " not found in protocol session");
- }
- this.channel = channel;
- this.protocolSession = protocolSession;
- this.consumerTag = consumerTag;
- _sessionKey = protocolSession.getKey();
- _acks = acks;
- _noLocal = noLocal;
- _queue = queue;
+ StoreContext storeContext = getChannel().getStoreContext();
+ try
+ { // if we do not need to wait for client acknowledgements
+ // we can decrement the reference count immediately.
- _filters = FilterManagerFactory.createManager(filters);
+ // By doing this _before_ the send we ensure that it
+ // doesn't get sent if it can't be dequeued, preventing
+ // duplicate delivery on recovery.
+ // The send may of course still fail, in which case, as
+ // the message is unacked, it will be lost.
+ entry.dequeue(storeContext);
- if (_filters != null)
- {
- Object isBrowser = filters.get(AMQPFilterTypes.NO_CONSUME.getValue());
- if (isBrowser != null)
- {
- _isBrowser = (Boolean) isBrowser;
- }
- else
- {
- _isBrowser = false;
- }
- }
- else
- {
- _isBrowser = false;
- }
+ synchronized (getChannel())
+ {
+ long deliveryTag = getChannel().getNextDeliveryTag();
+
+ getProtocolSession().getProtocolOutputConverter().writeDeliver(entry.getMessage(), getChannel().getChannelId(), deliveryTag, getConumerTag());
- if (_filters != null)
- {
- Object autoClose = filters.get(AMQPFilterTypes.AUTO_CLOSE.getValue());
- if (autoClose != null)
- {
- _autoClose = (Boolean) autoClose;
+ }
+ entry.dispose(storeContext);
}
- else
+ finally
{
- _autoClose = false;
+ //Only set delivered if it actually was writen successfully..
+ // using a try->finally would set it even if an error occured.
+ // Is this what we want?
+
+ entry.setDeliveredToSubscription();
}
}
- else
+
+ public boolean wouldSuspend(QueueEntry msg)
{
- _autoClose = false;
+ return false;
}
+ }
- if (filtersMessages())
+ static final class AckSubscription extends SubscriptionImpl
+ {
+ public AckSubscription(int channelId, AMQProtocolSession protocolSession,
+ AMQShortString consumerTag, FieldTable filters,
+ boolean noLocal, FlowCreditManager creditManager)
+ throws AMQException
{
- _messages = new ConcurrentLinkedMessageQueueAtomicSize<QueueEntry>();
+ super(channelId, protocolSession, consumerTag, filters, noLocal, creditManager);
}
- else
+
+ public boolean isBrowser()
{
- // Reference the DeliveryManager
- _messages = null;
+ return false;
}
- }
- public SubscriptionImpl(int channel, AMQProtocolSession protocolSession,
- AMQShortString consumerTag)
- throws AMQException
- {
- this(channel, protocolSession, consumerTag, false);
- }
+ /**
+ * This method can be called by each of the publisher threads. As a result all changes to the channel object must be
+ * thread safe.
+ *
+ * @param entry The message to send
+ * @throws AMQException
+ */
+ public void send(QueueEntry entry) throws AMQException
+ {
+ StoreContext storeContext = getChannel().getStoreContext();
+ try
+ { // if we do not need to wait for client acknowledgements
+ // we can decrement the reference count immediately.
- public boolean equals(Object o)
- {
- return (o instanceof SubscriptionImpl) && equals((SubscriptionImpl) o);
- }
+ // By doing this _before_ the send we ensure that it
+ // doesn't get sent if it can't be dequeued, preventing
+ // duplicate delivery on recovery.
- /**
- * Equality holds if the session matches and the channel and consumer tag are the same.
- *
- * @param psc The subscriptionImpl to compare
- *
- * @return equality
- */
- private boolean equals(SubscriptionImpl psc)
- {
- return _sessionKey.equals(psc._sessionKey)
- && psc.channel == channel
- && psc.consumerTag.equals(consumerTag);
- }
+ // The send may of course still fail, in which case, as
+ // the message is unacked, it will be lost.
- public int hashCode()
- {
- return _sessionKey.hashCode();
- }
+ synchronized (getChannel())
+ {
+ long deliveryTag = getChannel().getNextDeliveryTag();
- public String toString()
- {
- String subscriber = "[channel=" + channel +
- ", consumerTag=" + consumerTag +
- ", session=" + protocolSession.getKey() +
- ", resendQueue=" + (_resendQueue != null);
- if (_resendQueue != null)
- {
- subscriber += ", resendSize=" + _resendQueue.size();
- }
- return subscriber + "]";
- }
+ getChannel().addUnacknowledgedMessage(entry, deliveryTag, this);
- /**
- * This method can be called by each of the publisher threads. As a result all changes to the channel object must be
- * thread safe.
- *
- * @param msg The message to send
- * @param queue the Queue it has been sent from
- *
- * @throws AMQException
- */
- public void send(QueueEntry msg, AMQQueue queue) throws AMQException
- {
- if (msg != null)
- {
- if (_isBrowser)
- {
- sendToBrowser(msg, queue);
+ getProtocolSession().getProtocolOutputConverter().writeDeliver(entry.getMessage(), getChannel().getChannelId(), deliveryTag, getConumerTag());
+
+ }
}
- else
+ finally
{
- sendToConsumer(channel.getStoreContext(), msg, queue);
+ //Only set delivered if it actually was writen successfully..
+ // using a try->finally would set it even if an error occured.
+ // Is this what we want?
+
+ entry.setDeliveredToSubscription();
}
}
- else
- {
- _logger.error("Attempt to send Null message", new NullPointerException());
- }
+
+
+
}
- private void sendToBrowser(QueueEntry msg, AMQQueue queue) throws AMQException
- {
- // We don't decrement the reference here as we don't want to consume the message
- // but we do want to send it to the client.
- synchronized (channel)
- {
- long deliveryTag = channel.getNextDeliveryTag();
+ private static final Logger _logger = Logger.getLogger(SubscriptionImpl.class);
- // We don't need to add the message to the unacknowledgedMap as we don't need to know if the client
- // received the message. If it is lost in transit that is not important.
-// if (_acks)
-// {
-// channel.addUnacknowledgedBrowsedMessage(msg, deliveryTag, consumerTag, queue);
-// }
+ private final AMQChannel _channel;
- if (_sendLock.get())
- {
- _logger.error("Sending " + msg + " when subscriber(" + this + ") is closed!");
- }
+ private final AMQShortString _consumerTag;
- protocolSession.getProtocolOutputConverter().writeDeliver(msg.getMessage(), channel.getChannelId(), deliveryTag, consumerTag);
- }
- }
- private void sendToConsumer(StoreContext storeContext, QueueEntry entry, AMQQueue queue)
+ private final boolean _noLocal;
+
+ private final FlowCreditManager _creditManager;
+
+ private FilterManager _filters;
+
+ private final Boolean _autoClose;
+
+
+ private static final String CLIENT_PROPERTIES_INSTANCE = ClientProperties.instance.toString();
+
+ private AMQQueue _queue;
+ private final AtomicBoolean _sendLock = new AtomicBoolean(false);
+
+
+
+ public SubscriptionImpl(int channelId, AMQProtocolSession protocolSession,
+ AMQShortString consumerTag, FieldTable arguments,
+ boolean noLocal, FlowCreditManager creditManager)
throws AMQException
{
- try
- { // if we do not need to wait for client acknowledgements
- // we can decrement the reference count immediately.
-
- // By doing this _before_ the send we ensure that it
- // doesn't get sent if it can't be dequeued, preventing
- // duplicate delivery on recovery.
-
- // The send may of course still fail, in which case, as
- // the message is unacked, it will be lost.
- if (!_acks)
- {
- if (_logger.isDebugEnabled())
- {
- _logger.debug("No ack mode so dequeuing message immediately: " + entry.getMessage().getMessageId());
- }
- queue.dequeue(storeContext, entry);
- }
+ AMQChannel channel = protocolSession.getChannel(channelId);
+ if (channel == null)
+ {
+ throw new AMQException(AMQConstant.NOT_FOUND, "channel :" + channelId + " not found in protocol session");
+ }
-/*
- if (_sendLock.get())
- {
- _logger.error("Sending " + entry + " when subscriber(" + this + ") is closed!");
- }
-*/
+ _channel = channel;
+ _consumerTag = consumerTag;
- synchronized (channel)
- {
- long deliveryTag = channel.getNextDeliveryTag();
+ _creditManager = creditManager;
+ creditManager.addStateListener(this);
+
+ _noLocal = noLocal;
+
+
+ _filters = FilterManagerFactory.createManager(arguments);
- if (_acks)
- {
- channel.addUnacknowledgedMessage(entry, deliveryTag, consumerTag);
- }
- protocolSession.getProtocolOutputConverter().writeDeliver(entry.getMessage(), channel.getChannelId(), deliveryTag, consumerTag);
+ if (_filters != null)
+ {
+ Object autoClose = arguments.get(AMQPFilterTypes.AUTO_CLOSE.getValue());
+ if (autoClose != null)
+ {
+ _autoClose = (Boolean) autoClose;
}
- if (!_acks)
+ else
{
- entry.getMessage().decrementReference(storeContext);
+ _autoClose = false;
}
}
- finally
+ else
{
- //Only set delivered if it actually was writen successfully..
- // using a try->finally would set it even if an error occured.
- // Is this what we want?
+ _autoClose = false;
+ }
+
+
+ }
+
- entry.setDeliveredToConsumer();
+
+ public synchronized void setQueue(AMQQueue queue)
+ {
+ if(getQueue() != null)
+ {
+ throw new IllegalStateException("Attempt to set queue for subscription " + this + " to " + queue + "when already set to " + getQueue());
}
+ _queue = queue;
+ }
+
+ public String toString()
+ {
+ String subscriber = "[channel=" + _channel +
+ ", consumerTag=" + _consumerTag +
+ ", session=" + getProtocolSession().getKey() ;
+
+ return subscriber + "]";
}
+ /**
+ * This method can be called by each of the publisher threads. As a result all changes to the channel object must be
+ * thread safe.
+ *
+ * @param msg The message to send
+ * @throws AMQException
+ */
+ abstract public void send(QueueEntry msg) throws AMQException;
+
+
public boolean isSuspended()
{
-// if (_suspensionlogger.isInfoEnabled())
-// {
-// if (channel.isSuspended())
-// {
-// _suspensionlogger.debug("Subscription(" + debugIdentity() + ") channel's is susupended");
-// }
-// if (_sendLock.get())
-// {
-// _suspensionlogger.debug("Subscription(" + debugIdentity() + ") has sendLock set so closing.");
-// }
-// }
- return channel.isSuspended() || _sendLock.get();
+ return !isActive() || _channel.isSuspended() || _sendLock.get();
}
/**
@@ -349,9 +338,10 @@
*
* @param queue The queue to delete
*/
- public void queueDeleted(AMQQueue queue) throws AMQException
+ public void queueDeleted(AMQQueue queue)
{
- channel.queueDeleted(queue);
+ _sendLock.set(true);
+// _channel.queueDeleted(queue);
}
public boolean filtersMessages()
@@ -376,55 +366,40 @@
//todo - client id should be recoreded and this test removed but handled below
if (_noLocal)
{
+ final Object publisherId = entry.getMessage().getPublisherClientInstance();
- final AMQProtocolSession publisher = entry.getMessage().getPublisher();
- if(publisher != null)
+ // We don't want local messages so check to see if message is one we sent
+ Object localInstance;
+ if (publisherId != null && (getProtocolSession().getClientProperties() != null) &&
+ (localInstance = getProtocolSession().getClientProperties().getObject(CLIENT_PROPERTIES_INSTANCE)) != null)
{
- // We don't want local messages so check to see if message is one we sent
- Object localInstance;
- Object msgInstance;
-
- if ((protocolSession.getClientProperties() != null) &&
- (localInstance = protocolSession.getClientProperties().getObject(CLIENT_PROPERTIES_INSTANCE)) != null)
+ if(publisherId.equals(localInstance))
{
-
- if ((publisher.getClientProperties() != null) &&
- (msgInstance = publisher.getClientProperties().getObject(CLIENT_PROPERTIES_INSTANCE)) != null)
- {
- if (localInstance == msgInstance || localInstance.equals(msgInstance))
- {
- // if (_logger.isTraceEnabled())
- // {
- // _logger.trace("(" + debugIdentity() + ") has no interest as it is a local message(" +
- // msg.debugIdentity() + ")");
- // }
- return false;
- }
- }
+ return false;
}
- else
- {
+ }
+ else
+ {
- localInstance = protocolSession.getClientIdentifier();
- //todo - client id should be recoreded and this test removed but handled here
+ localInstance = getProtocolSession().getClientIdentifier();
+ //todo - client id should be recoreded and this test removed but handled here
- msgInstance = publisher.getClientIdentifier();
- if (localInstance == msgInstance || ((localInstance != null) && localInstance.equals(msgInstance)))
- {
- // if (_logger.isTraceEnabled())
- // {
- // _logger.trace("(" + debugIdentity() + ") has no interest as it is a local message(" +
- // msg.debugIdentity() + ")");
- // }
- return false;
- }
- }
+ if (localInstance != null && localInstance.equals(entry.getMessage().getPublisherIdentifier()))
+ {
+ return false;
+ }
}
+
+
}
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("(" + debugIdentity() + ") checking filters for message (" + entry.debugIdentity());
+ }
return checkFilters(entry);
}
@@ -441,27 +416,7 @@
return (_filters == null) || _filters.allAllow(msg.getMessage());
}
- public Queue<QueueEntry> getPreDeliveryQueue()
- {
- return _messages;
- }
-
- public void enqueueForPreDelivery(QueueEntry msg, boolean deliverFirst)
- {
- if (_messages != null)
- {
- if (deliverFirst)
- {
- _messages.pushHead(msg);
- }
- else
- {
- _messages.offer(msg);
- }
- }
- }
-
- private boolean isAutoClose()
+ public boolean isAutoClose()
{
return _autoClose;
}
@@ -469,14 +424,22 @@
public void close()
{
boolean closed = false;
+ State state = getState();
synchronized (_sendLock)
{
- if (_logger.isDebugEnabled())
+ while(!closed && state != State.CLOSED)
{
- _logger.debug("Setting SendLock true:" + debugIdentity());
+ closed = _state.compareAndSet(state, State.CLOSED);
+ if(!closed)
+ {
+ state = getState();
+ }
+ else
+ {
+ _stateListener.stateChange(this,state, State.CLOSED);
+ }
}
-
- closed = _sendLock.getAndSet(true);
+ _creditManager.removeListener(this);
}
if (closed)
@@ -493,197 +456,99 @@
{
_logger.info("Closing subscription (" + debugIdentity() + "):" + this);
}
-
- if (_resendQueue != null && !_resendQueue.isEmpty())
- {
- if (_logger.isInfoEnabled())
- {
- _logger.info("Requeuing closing subscription (" + debugIdentity() + "):" + this);
- }
- requeue();
- }
-
- //remove references in PDQ
- if (_messages != null)
- {
- if (_logger.isInfoEnabled())
- {
- _logger.info("Clearing PDQ (" + debugIdentity() + "):" + this);
- }
-
- _messages.clear();
- }
}
- private void autoclose()
+ public boolean isClosed()
{
- close();
-
- if (_autoClose && !_sentClose)
- {
- _logger.info("Closing autoclose subscription (" + debugIdentity() + "):" + this);
-
- boolean unregisteredOK = false;
- try
- {
- unregisteredOK = channel.unsubscribeConsumer(protocolSession, consumerTag);
- }
- catch (AMQException e)
- {
- // Occurs if we cannot find the subscriber in the channel with protocolSession and consumerTag.
- _logger.info("Unable to UnsubscribeConsumer :" + consumerTag +" so not going to send CancelOK.");
- }
+ return getState() == State.CLOSED;
+ }
- if (unregisteredOK)
- {
- ProtocolOutputConverter converter = protocolSession.getProtocolOutputConverter();
- converter.confirmConsumerAutoClose(channel.getChannelId(), consumerTag);
- _sentClose = true;
- }
- }
+ public boolean wouldSuspend(QueueEntry msg)
+ {
+ return !_creditManager.useCreditForMessage(msg.getMessage());//_channel.wouldSuspend(msg.getMessage());
}
- private void requeue()
+ public Object getSendLock()
{
- if (_queue != null)
- {
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Requeuing :" + _resendQueue.size() + " messages");
- }
-
- while (!_resendQueue.isEmpty())
- {
- QueueEntry resent = _resendQueue.poll();
-
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Removed for resending:" + resent.debugIdentity());
- }
-
- resent.release();
- _queue.subscriberHasPendingResend(false, this, resent);
-
- try
- {
- channel.getTransactionalContext().deliver(resent, true);
- }
- catch (AMQException e)
- {
- _logger.error("MESSAGE LOSS : Unable to re-deliver messages", e);
- }
- }
-
- if (!_resendQueue.isEmpty())
- {
- _logger.error("[MESSAGES LOST]Unable to re-deliver messages as queue is null.");
- }
-
- _queue.subscriberHasPendingResend(false, this, null);
- }
- else
- {
- if (!_resendQueue.isEmpty())
- {
- _logger.error("Unable to re-deliver messages as queue is null.");
- }
- }
+ return _sendLock;
+ }
- // Clear the messages
- _resendQueue = null;
+ public void resend(final QueueEntry entry) throws AMQException
+ {
+ _queue.resend(entry, this);
}
+ public AMQChannel getChannel()
+ {
+ return _channel;
+ }
- public boolean isClosed()
+ public AMQShortString getConumerTag()
{
- return _sendLock.get(); // This rather than _close is used to signify the subscriber is now closed.
+ return _consumerTag;
}
- public boolean isBrowser()
+ public AMQProtocolSession getProtocolSession()
{
- return _isBrowser;
+ return _channel.getProtocolSession();
}
- public boolean wouldSuspend(QueueEntry msg)
+ public AMQQueue getQueue()
{
- return _acks && channel.wouldSuspend(msg.getMessage());
+ return _queue;
}
- public Queue<QueueEntry> getResendQueue()
+ public void restoreCredit(final QueueEntry queueEntry)
{
- if (_resendQueue == null)
- {
- _resendQueue = new ConcurrentLinkedQueueAtomicSize<QueueEntry>();
- }
- return _resendQueue;
+ _creditManager.addCredit(1, queueEntry.getSize());
}
- public Queue<QueueEntry> getNextQueue(Queue<QueueEntry> messages)
+ public void creditStateChanged(boolean hasCredit)
{
- if (_resendQueue != null && !_resendQueue.isEmpty())
+
+ if(hasCredit)
{
- return _resendQueue;
- }
-
- if (filtersMessages())
- {
- if (isAutoClose())
+ if(_state.compareAndSet(State.SUSPENDED, State.ACTIVE))
{
- if (_messages.isEmpty())
- {
- autoclose();
- return null;
- }
+ _stateListener.stateChange(this, State.SUSPENDED, State.ACTIVE);
}
- return _messages;
}
- else // we want the DM queue
+ else
{
- return messages;
+ if(_state.compareAndSet(State.ACTIVE, State.SUSPENDED))
+ {
+ _stateListener.stateChange(this, State.ACTIVE, State.SUSPENDED);
+ }
}
}
- public void addToResendQueue(QueueEntry msg)
+ public State getState()
{
- // add to our resend queue
- getResendQueue().add(msg);
-
- // Mark Queue has having content.
- if (_queue == null)
- {
- _logger.error("Queue is null won't be able to resend messages");
- }
- else
- {
- _queue.subscriberHasPendingResend(true, this, msg);
- }
+ return _state.get();
}
- public Object getSendLock()
+
+ public void setStateListener(final StateListener listener)
{
- return _sendLock;
+ _stateListener = listener;
}
- public AMQChannel getChannel()
+
+ public Object getQueueContext()
{
- return channel;
+ return _queueContext.get();
}
- public void start()
+ public boolean setQueueContext(Object expected, Object newvalue)
{
- //Check to see if we need to autoclose
- if (filtersMessages())
- {
- if (isAutoClose())
- {
- if (_messages.isEmpty())
- {
- autoclose();
- }
- }
- }
+ return _queueContext.compareAndSet(expected,newvalue);
}
+
+ public boolean isActive()
+ {
+ return getState() == State.ACTIVE;
+ }
}
Propchange: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java?rev=650148&view=auto
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java (added)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java Mon Apr 21 07:03:35 2008
@@ -0,0 +1,247 @@
+/*
+*
+* 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.subscription;
+
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.subscription.Subscription;
+
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.nio.ByteBuffer;
+
+public class SubscriptionList
+{
+
+ private final SubscriptionNode _head = new SubscriptionNode();
+
+ private AtomicReference<SubscriptionNode> _tail = new AtomicReference<SubscriptionNode>(_head);
+ private final AMQQueue _queue;
+ private AtomicInteger _size = new AtomicInteger();
+
+
+ public final class SubscriptionNode
+ {
+ private final AtomicBoolean _deleted = new AtomicBoolean();
+ private final AtomicReference<SubscriptionNode> _next = new AtomicReference<SubscriptionNode>();
+ private final Subscription _sub;
+
+
+ public SubscriptionNode()
+ {
+
+ _sub = null;
+ _deleted.set(true);
+ }
+
+ public SubscriptionNode(final Subscription sub)
+ {
+ _sub = sub;
+ }
+
+
+ public SubscriptionNode getNext()
+ {
+
+ SubscriptionNode next = nextNode();
+ while(next != null && next.isDeleted())
+ {
+
+ final SubscriptionNode newNext = next.nextNode();
+ if(newNext != null)
+ {
+ _next.compareAndSet(next, newNext);
+ next = nextNode();
+ }
+ else
+ {
+ next = null;
+ }
+
+ }
+ return next;
+ }
+
+ private SubscriptionNode nextNode()
+ {
+ return _next.get();
+ }
+
+ public boolean isDeleted()
+ {
+ return _deleted.get();
+ }
+
+
+ public boolean delete()
+ {
+ if(_deleted.compareAndSet(false,true))
+ {
+ _size.decrementAndGet();
+ advanceHead();
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+ }
+
+
+ public Subscription getSubscription()
+ {
+ return _sub;
+ }
+ }
+
+
+ public SubscriptionList(AMQQueue queue)
+ {
+ _queue = queue;
+ }
+
+ private void advanceHead()
+ {
+ SubscriptionNode head = _head.nextNode();
+ while(head._next.get() != null && head.isDeleted())
+ {
+
+ final SubscriptionNode newhead = head.nextNode();
+ if(newhead != null)
+ {
+ _head._next.compareAndSet(head, newhead);
+ }
+ head = _head.nextNode();
+ }
+ }
+
+
+ public SubscriptionNode add(Subscription sub)
+ {
+ SubscriptionNode node = new SubscriptionNode(sub);
+ for (;;)
+ {
+ SubscriptionNode tail = _tail.get();
+ SubscriptionNode next = tail.nextNode();
+ if (tail == _tail.get())
+ {
+ if (next == null)
+ {
+ if (tail._next.compareAndSet(null, node))
+ {
+ _tail.compareAndSet(tail, node);
+ _size.incrementAndGet();
+ return node;
+ }
+ }
+ else
+ {
+ _tail.compareAndSet(tail, next);
+ }
+ }
+ }
+
+ }
+
+ public boolean remove(Subscription sub)
+ {
+ SubscriptionNode node = _head.getNext();
+ while(node != null)
+ {
+ if(sub.equals(node._sub) && node.delete())
+ {
+ return true;
+ }
+ node = node.getNext();
+ }
+ return false;
+ }
+
+
+ public class SubscriptionNodeIterator
+ {
+
+ private SubscriptionNode _lastNode;
+
+ SubscriptionNodeIterator(SubscriptionNode startNode)
+ {
+ _lastNode = startNode;
+ }
+
+
+ public boolean atTail()
+ {
+ return _lastNode.nextNode() == null;
+ }
+
+ public SubscriptionNode getNode()
+ {
+
+ return _lastNode;
+
+ }
+
+ public boolean advance()
+ {
+
+ if(!atTail())
+ {
+ SubscriptionNode nextNode = _lastNode.nextNode();
+ while(nextNode.isDeleted() && nextNode.nextNode() != null)
+ {
+ nextNode = nextNode.nextNode();
+ }
+ _lastNode = nextNode;
+ return true;
+
+ }
+ else
+ {
+ return false;
+ }
+
+ }
+
+ }
+
+
+ public SubscriptionNodeIterator iterator()
+ {
+ return new SubscriptionNodeIterator(_head);
+ }
+
+
+ public SubscriptionNode getHead()
+ {
+ return _head;
+ }
+
+ public int size()
+ {
+ return _size.get();
+ }
+
+
+
+}
+
+
+
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java Mon Apr 21 07:03:35 2008
@@ -24,18 +24,16 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.server.RequiredDeliveryException;
+import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.ack.TxAck;
import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.server.queue.AMQMessage;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.NoConsumersException;
-import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.queue.*;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.StoreContext;
-import java.util.LinkedList;
import java.util.List;
+import java.util.ArrayList;
/** A transactional context that only supports local transactions. */
public class LocalTransactionalContext implements TransactionalContext
@@ -44,7 +42,7 @@
private final TxnBuffer _txnBuffer = new TxnBuffer();
- private final List<DeliveryDetails> _postCommitDeliveryList = new LinkedList<DeliveryDetails>();
+ private final List<DeliveryAction> _postCommitDeliveryList = new ArrayList<DeliveryAction>();
/**
* We keep hold of the ack operation so that we can consolidate acks, i.e. multiple acks within a txn are
@@ -52,81 +50,112 @@
*/
private TxAck _ackOp;
- private List<RequiredDeliveryException> _returnMessages;
-
- private final MessageStore _messageStore;
-
- private final StoreContext _storeContext;
-
private boolean _inTran = false;
/** Are there messages to deliver. NOT Has the message been delivered */
private boolean _messageDelivered = false;
+ private final AMQChannel _channel;
- private static class DeliveryDetails
+
+ private abstract class DeliveryAction
{
- public QueueEntry entry;
- private boolean deliverFirst;
+ abstract public void process() throws AMQException;
- public DeliveryDetails(QueueEntry entry, boolean deliverFirst)
+ }
+
+ private class RequeueAction extends DeliveryAction
+ {
+ public QueueEntry entry;
+
+ public RequeueAction(QueueEntry entry)
{
this.entry = entry;
- this.deliverFirst = deliverFirst;
+ }
+
+ public void process() throws AMQException
+ {
+ entry.requeue(getStoreContext());
+ }
+ }
+
+ private class PublishAction extends DeliveryAction
+ {
+ private final AMQQueue _queue;
+ private final AMQMessage _message;
+
+ public PublishAction(final AMQQueue queue, final AMQMessage message)
+ {
+ _queue = queue;
+ _message = message;
+ }
+
+ public void process() throws AMQException
+ {
+
+ QueueEntry entry = _queue.enqueue(getStoreContext(),_message);
+
+ if(entry.immediateAndNotDelivered())
+ {
+ getReturnMessages().add(new NoConsumersException(_message));
+ }
}
}
- public LocalTransactionalContext(MessageStore messageStore, StoreContext storeContext,
- List<RequiredDeliveryException> returnMessages)
+ public LocalTransactionalContext(final AMQChannel channel)
{
- _messageStore = messageStore;
- _storeContext = storeContext;
- _returnMessages = returnMessages;
- // _txnBuffer.enlist(new StoreMessageOperation(messageStore));
+ _channel = channel;
}
public StoreContext getStoreContext()
{
- return _storeContext;
+ return _channel.getStoreContext();
+ }
+
+ public List<RequiredDeliveryException> getReturnMessages()
+ {
+ return _channel.getReturnMessages();
+ }
+
+ public MessageStore getMessageStore()
+ {
+ return _channel.getMessageStore();
}
+
public void rollback() throws AMQException
{
- _txnBuffer.rollback(_storeContext);
+ _txnBuffer.rollback(getStoreContext());
// Hack to deal with uncommitted non-transactional writes
- if (_messageStore.inTran(_storeContext))
+ if (getMessageStore().inTran(getStoreContext()))
{
- _messageStore.abortTran(_storeContext);
+ getMessageStore().abortTran(getStoreContext());
_inTran = false;
}
_postCommitDeliveryList.clear();
}
- public void deliver(QueueEntry entry, boolean deliverFirst) throws AMQException
+ public void deliver(final AMQQueue queue, AMQMessage message) throws AMQException
{
// A publication will result in the enlisting of several
// TxnOps. The first is an op that will store the message.
// Following that (and ordering is important), an op will
// be added for every queue onto which the message is
- // enqueued. Finally a cleanup op will be added to decrement
- // the reference associated with the routing.
- // message.incrementReference();
- _postCommitDeliveryList.add(new DeliveryDetails(entry, deliverFirst));
+ // enqueued.
+ _postCommitDeliveryList.add(new PublishAction(queue, message));
_messageDelivered = true;
- _txnBuffer.enlist(new CleanupMessageOperation(entry.getMessage(), _returnMessages));
- /*_txnBuffer.enlist(new DeliverMessageOperation(message, queue));
- if (_log.isDebugEnabled())
- {
- _log.debug("Incrementing ref count on message and enlisting cleanup operation - id " +
- message.getMessageId());
- }
- message.incrementReference();
+
+ }
+
+ public void requeue(QueueEntry entry) throws AMQException
+ {
+ _postCommitDeliveryList.add(new RequeueAction(entry));
_messageDelivered = true;
- */
}
+
private void checkAck(long deliveryTag, UnacknowledgedMessageMap unacknowledgedMessageMap) throws AMQException
{
if (!unacknowledgedMessageMap.contains(deliveryTag))
@@ -184,7 +213,7 @@
_log.debug("Starting transaction on message store: " + this);
}
- _messageStore.beginTran(_storeContext);
+ getMessageStore().beginTran(getStoreContext());
_inTran = true;
}
}
@@ -207,22 +236,22 @@
if (_messageDelivered && _inTran)
{
- _txnBuffer.enlist(new StoreMessageOperation(_messageStore));
+ _txnBuffer.enlist(new StoreMessageOperation(getMessageStore()));
}
// fixme fail commit here ... QPID-440
try
{
- _txnBuffer.commit(_storeContext);
+ _txnBuffer.commit(getStoreContext());
}
finally
{
_messageDelivered = false;
- _inTran = _messageStore.inTran(_storeContext);
+ _inTran = getMessageStore().inTran(getStoreContext());
}
try
{
- postCommitDelivery(_returnMessages);
+ postCommitDelivery();
}
catch (AMQException e)
{
@@ -231,7 +260,7 @@
}
}
- private void postCommitDelivery(List<RequiredDeliveryException> returnMessages) throws AMQException
+ private void postCommitDelivery() throws AMQException
{
if (_log.isDebugEnabled())
{
@@ -240,18 +269,9 @@
try
{
- for (DeliveryDetails dd : _postCommitDeliveryList)
+ for (DeliveryAction dd : _postCommitDeliveryList)
{
- dd.entry.process(_storeContext, dd.deliverFirst);
-
- try
- {
- dd.entry.checkDeliveredToConsumer();
- }
- catch (NoConsumersException nce)
- {
- returnMessages.add(nce);
- }
+ dd.process();
}
}
finally
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java Mon Apr 21 07:03:35 2008
@@ -22,19 +22,14 @@
import java.util.LinkedList;
import java.util.List;
-import java.util.Set;
import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.RequiredDeliveryException;
-import org.apache.qpid.server.ack.UnacknowledgedMessage;
import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.server.queue.AMQMessage;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.NoConsumersException;
-import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.queue.*;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.StoreContext;
@@ -49,7 +44,7 @@
/** Where to put undeliverable messages */
private final List<RequiredDeliveryException> _returnMessages;
- private final Set<Long> _browsedAcks;
+
private final MessageStore _messageStore;
@@ -59,13 +54,13 @@
private boolean _inTran;
public NonTransactionalContext(MessageStore messageStore, StoreContext storeContext, AMQChannel channel,
- List<RequiredDeliveryException> returnMessages, Set<Long> browsedAcks)
+ List<RequiredDeliveryException> returnMessages)
{
_channel = channel;
_storeContext = storeContext;
_returnMessages = returnMessages;
_messageStore = messageStore;
- _browsedAcks = browsedAcks;
+
}
@@ -93,19 +88,22 @@
// Does not apply to this context
}
- public void deliver(QueueEntry entry, boolean deliverFirst) throws AMQException
+ public void deliver(final AMQQueue queue, AMQMessage message) throws AMQException
{
- try
- {
- entry.process(_storeContext, deliverFirst);
- //following check implements the functionality
- //required by the 'immediate' flag:
- entry.checkDeliveredToConsumer();
- }
- catch (NoConsumersException e)
+ QueueEntry entry = queue.enqueue(_storeContext, message);
+
+ //following check implements the functionality
+ //required by the 'immediate' flag:
+ if(entry.immediateAndNotDelivered())
{
- _returnMessages.add(e);
+ _returnMessages.add(new NoConsumersException(entry.getMessage()));
}
+
+ }
+
+ public void requeue(QueueEntry entry) throws AMQException
+ {
+ entry.requeue(_storeContext);
}
public void acknowledgeMessage(final long deliveryTag, long lastDeliveryTag,
@@ -123,22 +121,17 @@
unacknowledgedMessageMap.size());
unacknowledgedMessageMap.visit(new UnacknowledgedMessageMap.Visitor()
{
- public boolean callback(UnacknowledgedMessage message) throws AMQException
+ public boolean callback(final long deliveryTag, QueueEntry message) throws AMQException
{
- if (!_browsedAcks.contains(deliveryTag))
- {
- if (_log.isDebugEnabled())
- {
- _log.debug("Discarding message: " + message.getMessage().getMessageId());
- }
-
- //Message has been ack so discard it. This will dequeue and decrement the reference.
- message.discard(_storeContext);
- }
- else
+ if (_log.isDebugEnabled())
{
- _browsedAcks.remove(deliveryTag);
+ _log.debug("Discarding message: " + message.getMessage().getMessageId());
}
+
+ message.restoreCredit();
+ //Message has been ack so discard it. This will dequeue and decrement the reference.
+ message.discard(_storeContext);
+
return false;
}
@@ -155,30 +148,24 @@
throw new AMQException("Multiple ack on delivery tag " + deliveryTag + " not known for channel");
}
- LinkedList<UnacknowledgedMessage> acked = new LinkedList<UnacknowledgedMessage>();
+ LinkedList<QueueEntry> acked = new LinkedList<QueueEntry>();
unacknowledgedMessageMap.drainTo(acked, deliveryTag);
- for (UnacknowledgedMessage msg : acked)
+ for (QueueEntry msg : acked)
{
- if (!_browsedAcks.contains(deliveryTag))
- {
if (_log.isDebugEnabled())
{
_log.debug("Discarding message: " + msg.getMessage().getMessageId());
}
+
//Message has been ack so discard it. This will dequeue and decrement the reference.
msg.discard(_storeContext);
- }
- else
- {
- _browsedAcks.remove(deliveryTag);
- }
}
}
}
else
{
- UnacknowledgedMessage msg;
+ QueueEntry msg;
msg = unacknowledgedMessageMap.remove(deliveryTag);
if (msg == null)
@@ -189,21 +176,14 @@
_channel.getChannelId());
}
- if (!_browsedAcks.contains(deliveryTag))
- {
- if (_log.isDebugEnabled())
- {
- _log.debug("Discarding message: " + msg.getMessage().getMessageId());
- }
-
- //Message has been ack so discard it. This will dequeue and decrement the reference.
- msg.discard(_storeContext);
- }
- else
+ if (_log.isDebugEnabled())
{
- _browsedAcks.remove(deliveryTag);
+ _log.debug("Discarding message: " + msg.getMessage().getMessageId());
}
+ //Message has been ack so discard it. This will dequeue and decrement the reference.
+ msg.discard(_storeContext);
+
if (_log.isDebugEnabled())
{
_log.debug("Received non-multiple ack for messaging with delivery tag " + deliveryTag + " msg id " +
@@ -223,6 +203,6 @@
public void messageProcessed(AMQProtocolSession protocolSession) throws AMQException
{
- _channel.processReturns(protocolSession);
+ _channel.processReturns();
}
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java Mon Apr 21 07:03:35 2008
@@ -25,6 +25,7 @@
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.store.StoreContext;
/**
@@ -106,18 +107,26 @@
void rollback() throws AMQException;
/**
- * Delivers the specified message to the specified queue. A 'deliverFirst' flag may be set if the message is a
- * redelivery, and should be placed on the front of the queue.
+ * Delivers the specified message to the specified queue.
*
* <p/>This is an 'enqueue' operation.
*
- * @param entry The message to deliver, and the queue to deliver to.
- * @param deliverFirst <tt>true</tt> to place the message on the front of the queue for redelivery, <tt>false</tt>
- * for normal FIFO message ordering.
- *
+ * @param queue
+ * @param message The message to deliver
* @throws AMQException If the message cannot be delivered for any reason.
*/
- void deliver(QueueEntry entry, boolean deliverFirst) throws AMQException;
+ void deliver(final AMQQueue queue, AMQMessage message) throws AMQException;
+
+ /**
+ * Requeues the specified message entry (message queue pair)
+ *
+ *
+ * @param queueEntry The message,queue pair
+ *
+ * @throws AMQException If the message cannot be delivered for any reason.
+ */
+ void requeue(QueueEntry queueEntry) throws AMQException;
+
/**
* Acknowledges a message or many messages as delivered. All messages up to a specified one, may be acknowledged by
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/MessageStoreTool.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/MessageStoreTool.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/MessageStoreTool.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/MessageStoreTool.java Mon Apr 21 07:03:35 2008
@@ -25,11 +25,11 @@
import org.apache.commons.configuration.ConfigurationException;
import org.apache.qpid.configuration.Configuration;
import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry;
import org.apache.qpid.server.store.MemoryMessageStore;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.tools.messagestore.commands.Clear;
import org.apache.qpid.tools.messagestore.commands.Command;
import org.apache.qpid.tools.messagestore.commands.Copy;
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Copy.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Copy.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Copy.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Copy.java Mon Apr 21 07:03:35 2008
@@ -20,8 +20,8 @@
*/
package org.apache.qpid.tools.messagestore.commands;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.tools.messagestore.MessageStoreTool;
+import org.apache.qpid.server.queue.AMQQueue;
public class Copy extends Move
{
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Dump.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Dump.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Dump.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Dump.java Mon Apr 21 07:03:35 2008
@@ -24,6 +24,7 @@
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.framing.abstraction.ContentChunk;
import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.queue.QueueEntryImpl;
import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.tools.messagestore.MessageStoreTool;
import org.apache.qpid.tools.utils.Console;
@@ -255,7 +256,7 @@
String title, boolean routing, boolean headers, boolean messageHeaders)
{
List<QueueEntry> single = new LinkedList<QueueEntry>();
- single.add(new QueueEntry(null,msg));
+ single.add(new QueueEntryImpl(null,msg, Long.MIN_VALUE));
List<List> routingData = super.createMessageData(null, single, headers, routing, messageHeaders);
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/List.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/List.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/List.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/List.java Mon Apr 21 07:03:35 2008
@@ -22,6 +22,7 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.queue.AMQQueueImpl;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.virtualhost.VirtualHost;
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Move.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Move.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Move.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Move.java Mon Apr 21 07:03:35 2008
@@ -21,7 +21,7 @@
package org.apache.qpid.tools.messagestore.commands;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.queue.QueueEntryImpl;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.server.store.StoreContext;
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Purge.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Purge.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Purge.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Purge.java Mon Apr 21 07:03:35 2008
@@ -20,9 +20,8 @@
*/
package org.apache.qpid.tools.messagestore.commands;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.store.StoreContext;
import org.apache.qpid.tools.messagestore.MessageStoreTool;
+import org.apache.qpid.server.queue.AMQQueue;
public class Purge extends Move
{
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Select.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Select.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Select.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Select.java Mon Apr 21 07:03:35 2008
@@ -22,9 +22,9 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.tools.messagestore.MessageStoreTool;
import java.util.LinkedList;
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java Mon Apr 21 07:03:35 2008
@@ -26,6 +26,7 @@
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.queue.QueueEntryImpl;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.tools.messagestore.MessageStoreTool;
@@ -33,7 +34,6 @@
import java.util.LinkedList;
import java.util.List;
-import java.util.StringTokenizer;
public class Show extends AbstractCommand
{