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 2016/11/09 14:39:49 UTC
svn commit: r1768979 - in /qpid/java/branches/remove-queue-runner:
broker-core/src/main/java/org/apache/qpid/server/consumer/
broker-core/src/main/java/org/apache/qpid/server/flow/
broker-core/src/main/java/org/apache/qpid/server/queue/ broker-core/src...
Author: rgodfrey
Date: Wed Nov 9 14:39:48 2016
New Revision: 1768979
URL: http://svn.apache.org/viewvc?rev=1768979&view=rev
Log:
Have consumer targets keep the record of whether they desire notifyWork or not
Added:
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/InfiniteCreditCreditManager.java
- copied, changed from r1768978, qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageOnlyCreditManager.java
Removed:
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageOnlyCreditManager.java
Modified:
qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java
qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java
qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java
qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java
qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
qpid/java/branches/remove-queue-runner/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/AMQPConnection_0_10.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/FlowCreditManager_0_10.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageAcceptCompletionListener.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java
qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
Modified: qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java Wed Nov 9 14:39:48 2016
@@ -26,11 +26,8 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
@@ -55,15 +52,13 @@ public abstract class AbstractConsumerTa
private final Set<StateChangeListener<ConsumerTarget, State>> _stateChangeListeners = new
CopyOnWriteArraySet<>();
- private final Lock _stateChangeLock = new ReentrantLock();
private final AtomicInteger _stateActivates = new AtomicInteger();
private final boolean _isMultiQueue;
private final SuspendedConsumerLoggingTicker _suspendedConsumerLoggingTicker;
private final List<ConsumerImpl> _consumers = new CopyOnWriteArrayList<>();
private Iterator<ConsumerImpl> _pullIterator;
- private final AtomicBoolean _waitingOnStateChange = new AtomicBoolean();
-
+ private boolean _notifyWorkDesired;
protected AbstractConsumerTarget(final State initialState,
final boolean isMultiQueue,
@@ -92,10 +87,19 @@ public abstract class AbstractConsumerTa
@Override
public void notifyWork()
{
- _waitingOnStateChange.set(false);
getSessionModel().notifyWork(this);
}
+ protected final void setNotifyWorkDesired(final boolean desired)
+ {
+ _notifyWorkDesired = desired;
+ }
+
+ protected final boolean isNotifyWorkDesired()
+ {
+ return _notifyWorkDesired;
+ }
+
@Override
public boolean processPending()
{
@@ -103,26 +107,11 @@ public abstract class AbstractConsumerTa
{
return false;
}
+ // TODO - if not closed
return sendNextMessage();
}
@Override
- public boolean hasPendingWork()
- {
- if (!_waitingOnStateChange.get() && hasCredit())
- {
- for (ConsumerImpl consumer : _consumers)
- {
- if (consumer.hasAvailableMessages())
- {
- return true;
- }
- }
- }
- return false;
- }
-
- @Override
public void consumerAdded(final ConsumerImpl sub)
{
_consumers.add(sub);
@@ -290,8 +279,6 @@ public abstract class AbstractConsumerTa
@Override
public boolean sendNextMessage()
{
- _waitingOnStateChange.set(true);
-
AbstractQueue.MessageContainer messageContainer = null;
ConsumerImpl consumer = null;
boolean iteratedCompleteList = false;
@@ -316,7 +303,6 @@ public abstract class AbstractConsumerTa
if (messageContainer != null)
{
- _waitingOnStateChange.set(false);
MessageInstance entry = messageContainer._messageInstance;
try
{
Modified: qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java Wed Nov 9 14:39:48 2016
@@ -37,8 +37,6 @@ public interface ConsumerTarget
boolean processPending();
- boolean hasPendingWork();
-
String getTargetAddress();
boolean hasCredit();
@@ -47,6 +45,8 @@ public interface ConsumerTarget
void notifyWork();
+ void updateNotifyWorkDesired();
+
enum State
{
ACTIVE, SUSPENDED, CLOSED
@@ -74,8 +74,6 @@ public interface ConsumerTarget
void flushBatched();
- void queueDeleted();
-
void queueEmpty();
boolean allocateCredit(ServerMessage msg);
Modified: qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java Wed Nov 9 14:39:48 2016
@@ -24,7 +24,7 @@ package org.apache.qpid.server.flow;
public interface FlowCreditManager
{
- public static interface FlowCreditManagerListener
+ interface FlowCreditManagerListener
{
void creditStateChanged(boolean hasCredit);
}
@@ -33,10 +33,10 @@ public interface FlowCreditManager
boolean removeListener(FlowCreditManagerListener listener);
- public void restoreCredit(long messageCredit, long bytesCredit);
+ void restoreCredit(long messageCredit, long bytesCredit);
- public boolean hasCredit();
+ boolean hasCredit();
- public boolean useCreditForMessage(long msgSize);
+ boolean useCreditForMessage(long msgSize);
}
Modified: qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java Wed Nov 9 14:39:48 2016
@@ -2011,7 +2011,7 @@ public abstract class AbstractQueue<X ex
{
if (sub.hasInterest(node) && mightAssign(sub, node))
{
- if (!sub.wouldSuspend(node))
+ if (sub.allocateCredit(node))
{
MessageReference messageReference = null;
try
@@ -2038,13 +2038,9 @@ public abstract class AbstractQueue<X ex
}
}
}
- else // Not enough Credit for message and wouldSuspend
+ else
{
- //QPID-1187 - Treat the consumer as suspended for this message
- // and wait for the message to be removed to continue delivery.
- subActive = false;
sub.awaitCredit(node);
-
}
}
}
Modified: qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java Wed Nov 9 14:39:48 2016
@@ -32,7 +32,7 @@ public interface QueueConsumer<X extends
boolean hasInterest(QueueEntry node);
- boolean wouldSuspend(QueueEntry entry);
+ boolean allocateCredit(QueueEntry entry);
void restoreCredit(QueueEntry entry);
Modified: qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java Wed Nov 9 14:39:48 2016
@@ -352,12 +352,12 @@ class QueueConsumerImpl
public void queueDeleted()
{
- _target.queueDeleted();
+ _target.consumerRemoved(this);
}
- public boolean wouldSuspend(final QueueEntry msg)
+ public boolean allocateCredit(final QueueEntry msg)
{
- return !_target.allocateCredit(msg.getMessage());
+ return _target.allocateCredit(msg.getMessage());
}
public void restoreCredit(final QueueEntry queueEntry)
Modified: qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java Wed Nov 9 14:39:48 2016
@@ -91,4 +91,6 @@ public interface AMQPConnection<C extend
void resetStatistics();
void notifyWork(AMQSessionModel<?> sessionModel);
+
+ boolean isTransportBlockedForWriting();
}
Modified: qpid/java/branches/remove-queue-runner/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java Wed Nov 9 14:39:48 2016
@@ -135,10 +135,6 @@ public class TestConsumerTarget implemen
return false;
}
- public void queueDeleted()
- {
- }
-
public void restoreCredit(ServerMessage message)
{
}
@@ -246,12 +242,6 @@ public class TestConsumerTarget implemen
return true;
}
- @Override
- public boolean hasPendingWork()
- {
- return false;
- }
-
public ArrayList<MessageInstance> getMessages()
{
return _messages;
@@ -285,6 +275,12 @@ public class TestConsumerTarget implemen
{
}
+
+ @Override
+ public void updateNotifyWorkDesired()
+ {
+
+ }
private static class MockSessionModel implements AMQSessionModel<MockSessionModel>
{
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/AMQPConnection_0_10.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/AMQPConnection_0_10.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/AMQPConnection_0_10.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/AMQPConnection_0_10.java Wed Nov 9 14:39:48 2016
@@ -306,6 +306,7 @@ public class AMQPConnection_0_10 extends
public void sendConnectionCloseAsync(final AMQConstant cause, final String message)
{
+ stopConnection();
_connection.sendConnectionCloseAsync(cause, message);
}
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java Wed Nov 9 14:39:48 2016
@@ -68,7 +68,6 @@ public class ConsumerTarget_0_10 extends
private static final Option[] BATCHED = new Option[] { Option.BATCH };
- private final AtomicBoolean _deleted = new AtomicBoolean(false);
private final String _name;
private final String _targetAddress;
@@ -138,11 +137,22 @@ public class ConsumerTarget_0_10 extends
@Override
public boolean isFlowSuspended()
{
- return getState()!=State.ACTIVE || _deleted.get() || _session.isClosing() || _session.getAMQPConnection().isConnectionStopped();
+ return getState()!=State.ACTIVE || _session.isClosing() || _session.getAMQPConnection().isConnectionStopped();
// TODO check for Session suspension
}
@Override
+ public void updateNotifyWorkDesired()
+ {
+ final AMQPConnection_0_10 amqpConnection = _session.getAMQPConnection();
+
+ boolean state = !amqpConnection.isTransportBlockedForWriting()
+ && getCreditManager().hasCredit();
+
+ setNotifyWorkDesired(state);
+ }
+
+ @Override
protected void doCloseInternal()
{
_creditManager.removeListener(this);
@@ -173,6 +183,7 @@ public class ConsumerTarget_0_10 extends
public void transportStateChanged()
{
_creditManager.restoreCredit(0, 0);
+ updateNotifyWorkDesired();
}
public static class AddMessageDispositionListenerAction implements Runnable
@@ -401,7 +412,8 @@ public class ConsumerTarget_0_10 extends
|| !(_creditManager instanceof WindowCreditManager)
|| ((WindowCreditManager)_creditManager).getMessageCreditLimit() < 400 )
{
- _creditManager.restoreCredit(_deferredMessageCredit, _deferredSizeCredit);
+ restoreCredit(_deferredMessageCredit, _deferredSizeCredit);
+
_deferredMessageCredit = 0;
_deferredSizeCredit = 0l;
}
@@ -514,21 +526,34 @@ public class ConsumerTarget_0_10 extends
return (maxDeliveryLimit > 0 && entry.getDeliveryCount() >= maxDeliveryLimit);
}
- public void queueDeleted()
+ public boolean allocateCredit(ServerMessage message)
{
- _deleted.set(true);
+ boolean hasCredit = _creditManager.hasCredit();
+ boolean creditAllocated = _creditManager.useCreditForMessage(message.getSize());
+ if(_creditManager.hasCredit() != hasCredit)
+ {
+ updateNotifyWorkDesired();
+ }
+ return creditAllocated;
}
- public boolean allocateCredit(ServerMessage message)
+ public void restoreCredit(ServerMessage message)
{
- return _creditManager.useCreditForMessage(message.getSize());
+ restoreCredit(1, message.getSize());
}
- public void restoreCredit(ServerMessage message)
+ void restoreCredit(int count, long size)
{
- _creditManager.restoreCredit(1, message.getSize());
+ boolean hasCredit = _creditManager.hasCredit();
+ _creditManager.restoreCredit(count, size);
+ if(_creditManager.hasCredit() != hasCredit)
+ {
+ updateNotifyWorkDesired();
+ }
}
+
+
public FlowCreditManager_0_10 getCreditManager()
{
return _creditManager;
@@ -539,13 +564,18 @@ public class ConsumerTarget_0_10 extends
updateState(State.ACTIVE, State.SUSPENDED);
_stopped.set(true);
FlowCreditManager_0_10 creditManager = getCreditManager();
+ boolean hasCredit = creditManager.hasCredit();
creditManager.clearCredit();
+ if(hasCredit)
+ {
+ updateNotifyWorkDesired();
+ }
}
public void addCredit(MessageCreditUnit unit, long value)
{
FlowCreditManager_0_10 creditManager = getCreditManager();
-
+ boolean hasCredit = creditManager.hasCredit();
switch (unit)
{
case MESSAGE:
@@ -559,10 +589,15 @@ public class ConsumerTarget_0_10 extends
_stopped.set(false);
- if(creditManager.hasCredit())
+ boolean newHasCredit = creditManager.hasCredit();
+ if(newHasCredit)
{
updateState(State.SUSPENDED, State.ACTIVE);
}
+ if(hasCredit != newHasCredit)
+ {
+ updateNotifyWorkDesired();
+ }
}
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/FlowCreditManager_0_10.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/FlowCreditManager_0_10.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/FlowCreditManager_0_10.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/FlowCreditManager_0_10.java Wed Nov 9 14:39:48 2016
@@ -24,7 +24,7 @@ import org.apache.qpid.server.flow.FlowC
public interface FlowCreditManager_0_10 extends FlowCreditManager
{
- public void addCredit(long count, long bytes);
+ void addCredit(long count, long bytes);
void clearCredit();
}
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageAcceptCompletionListener.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageAcceptCompletionListener.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageAcceptCompletionListener.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageAcceptCompletionListener.java Wed Nov 9 14:39:48 2016
@@ -56,7 +56,7 @@ public class MessageAcceptCompletionList
{
if(_restoreCredit)
{
- _sub.getCreditManager().restoreCredit(1l, _messageSize);
+ _sub.restoreCredit(1, _messageSize);
}
_session.acknowledge(_consumer, _sub, _entry);
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java Wed Nov 9 14:39:48 2016
@@ -509,7 +509,8 @@ public class ServerConnection extends Co
@Override
public boolean hasNext()
{
- return !(_sessionsWithPending.isEmpty() && _asyncTaskList.isEmpty());
+ return (!_sessionsWithPending.isEmpty() && !isClosing() && !_amqpConnection.isConnectionStopped())
+ || !_asyncTaskList.isEmpty();
}
@Override
@@ -517,23 +518,39 @@ public class ServerConnection extends Co
{
if(!_sessionsWithPending.isEmpty())
{
- if(!_sessionIterator.hasNext())
+ if(isClosing() || _amqpConnection.isConnectionStopped())
{
- _sessionIterator = _sessionsWithPending.iterator();
+ // in case the connection was marked as closing between a call to hasNext() and
+ // a subsequent call to next()
+ return new Runnable()
+ {
+ @Override
+ public void run()
+ {
+
+ }
+ };
}
- final AMQSessionModel<?> session = _sessionIterator.next();
- return new Runnable()
+ else
{
- @Override
- public void run()
+ if (!_sessionIterator.hasNext())
{
- _sessionIterator.remove();
- if(session.processPending())
+ _sessionIterator = _sessionsWithPending.iterator();
+ }
+ final AMQSessionModel<?> session = _sessionIterator.next();
+ return new Runnable()
+ {
+ @Override
+ public void run()
{
- _sessionsWithPending.add(session);
+ _sessionIterator.remove();
+ if (session.processPending())
+ {
+ _sessionsWithPending.add(session);
+ }
}
- }
- };
+ };
+ }
}
else if(!_asyncTaskList.isEmpty())
{
@@ -551,6 +568,7 @@ public class ServerConnection extends Co
{
throw new NoSuchElementException();
}
+
}
@Override
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java Wed Nov 9 14:39:48 2016
@@ -1188,7 +1188,7 @@ public class ServerSession extends Sessi
@Override
public boolean processPending()
{
- if (!getAMQPConnection().isIOThread())
+ if (!getAMQPConnection().isIOThread() || isClosing())
{
return false;
}
@@ -1209,15 +1209,14 @@ public class ServerSession extends Sessi
_blockTime = desiredBlockingState ? System.currentTimeMillis() : 0;
}
-
- if(!_consumersWithPendingWork.isEmpty())
+ if (!_consumersWithPendingWork.isEmpty())
{
if (_processPendingIterator == null || !_processPendingIterator.hasNext())
{
_processPendingIterator = _consumersWithPendingWork.iterator();
}
- if(_processPendingIterator.hasNext())
+ if (_processPendingIterator.hasNext())
{
ConsumerTarget target = _processPendingIterator.next();
_processPendingIterator.remove();
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java Wed Nov 9 14:39:48 2016
@@ -233,6 +233,7 @@ public class AMQChannel
private boolean _logChannelFlowMessages = true;
private final CachedFrame _txCommitOkFrame;
+ private boolean _channelFlow = true;
public AMQChannel(AMQPConnection_0_8 connection, int channelId, final MessageStore messageStore)
{
@@ -304,10 +305,10 @@ public class AMQChannel
MessageSource.ExistingExclusiveConsumer, MessageSource.ConsumerAccessRefused
{
- final FlowCreditManager singleMessageCredit = new MessageOnlyCreditManager(1L);
+ final FlowCreditManager singleMessageCredit = new InfiniteCreditCreditManager();
final GetDeliveryMethod getDeliveryMethod =
- new GetDeliveryMethod(singleMessageCredit, queue);
+ new GetDeliveryMethod(queue);
final RecordDeliveryMethod getRecordMethod = new RecordDeliveryMethod()
{
@@ -1456,6 +1457,11 @@ public class AMQChannel
return _maxUncommittedInMemorySize;
}
+ public boolean isChannelFlow()
+ {
+ return _channelFlow;
+ }
+
private class NoLocalFilter implements MessageFilter
{
@@ -1493,15 +1499,11 @@ public class AMQChannel
private class GetDeliveryMethod implements ClientDeliveryMethod
{
-
- private final FlowCreditManager _singleMessageCredit;
private final MessageSource _queue;
private boolean _deliveredMessage;
- public GetDeliveryMethod(final FlowCreditManager singleMessageCredit,
- final MessageSource queue)
+ public GetDeliveryMethod(final MessageSource queue)
{
- _singleMessageCredit = singleMessageCredit;
_queue = queue;
}
@@ -1509,7 +1511,7 @@ public class AMQChannel
public long deliverToClient(final ConsumerImpl sub, final ServerMessage message,
final InstanceProperties props, final long deliveryTag)
{
- _singleMessageCredit.useCreditForMessage(message.getSize());
+
int queueSize = _queue instanceof Queue ? ((Queue<?>)_queue).getQueueDepthMessages() : 0;
long size = _connection.getProtocolOutputConverter().writeGetOk(message,
props,
@@ -1750,10 +1752,19 @@ public class AMQChannel
@Override
public void transportStateChanged()
{
+ updateAllConsumerNotifyWorkDesired();
_creditManager.restoreCredit(0, 0);
_noAckCreditManager.restoreCredit(0, 0);
}
+ void updateAllConsumerNotifyWorkDesired()
+ {
+ for(ConsumerTarget_0_8 target : _tag2SubscriptionTargetMap.values())
+ {
+ target.updateNotifyWorkDesired();
+ }
+ }
+
@Override
public Object getConnectionReference()
{
@@ -1765,7 +1776,7 @@ public class AMQChannel
return getUnacknowledgedMessageMap().size();
}
- private void flow(boolean flow)
+ private void sendFlow(boolean flow)
{
MethodRegistry methodRegistry = _connection.getMethodRegistry();
AMQMethodBody responseBody = methodRegistry.createChannelFlowBody(flow);
@@ -2663,6 +2674,13 @@ public class AMQChannel
sync();
+ if(_channelFlow != active)
+ {
+ _channelFlow = active;
+ // inform consumer targets
+ updateAllConsumerNotifyWorkDesired();
+ }
+
setSuspended(!active);
MethodRegistry methodRegistry = _connection.getMethodRegistry();
@@ -3729,7 +3747,7 @@ public class AMQChannel
@Override
public boolean processPending()
{
- if (!getAMQPConnection().isIOThread())
+ if (!getAMQPConnection().isIOThread() || isClosing())
{
return false;
}
@@ -3738,7 +3756,7 @@ public class AMQChannel
if (desiredBlockingState != _wireBlockingState)
{
_wireBlockingState = desiredBlockingState;
- flow(!desiredBlockingState);
+ sendFlow(!desiredBlockingState);
_blockTime = desiredBlockingState ? System.currentTimeMillis() : 0;
}
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java Wed Nov 9 14:39:48 2016
@@ -801,6 +801,7 @@ public class AMQPConnection_0_8Impl
@Override
public void sendConnectionCloseAsync(final AMQConstant cause, final String message)
{
+ stopConnection();
Action<AMQPConnection_0_8Impl> action = new Action<AMQPConnection_0_8Impl>()
{
@Override
@@ -1436,7 +1437,7 @@ public class AMQPConnection_0_8Impl
@Override
public boolean hasNext()
{
- return !(_sessionsWithWork.isEmpty() && _asyncTaskList.isEmpty());
+ return (!_sessionsWithWork.isEmpty() && !isClosing() && !isConnectionStopped()) || !_asyncTaskList.isEmpty();
}
@Override
@@ -1444,24 +1445,40 @@ public class AMQPConnection_0_8Impl
{
if(!_sessionsWithWork.isEmpty())
{
- if(!_sessionIterator.hasNext())
+ if(isClosing() || isConnectionStopped())
{
- _sessionIterator = _sessionsWithWork.iterator();
+ // in case the connection was marked as closing between a call to hasNext() and
+ // a subsequent call to next()
+ return new Runnable()
+ {
+ @Override
+ public void run()
+ {
+
+ }
+ };
}
- final AMQSessionModel<?> session = _sessionIterator.next();
- return new Runnable()
+ else
{
- @Override
- public void run()
+ if (!_sessionIterator.hasNext())
{
- _sessionIterator.remove();
-
- if(session.processPending())
+ _sessionIterator = _sessionsWithWork.iterator();
+ }
+ final AMQSessionModel<?> session = _sessionIterator.next();
+ return new Runnable()
+ {
+ @Override
+ public void run()
{
- _sessionsWithWork.add(session);
+ _sessionIterator.remove();
+
+ if (session.processPending())
+ {
+ _sessionsWithWork.add(session);
+ }
}
- }
- };
+ };
+ }
}
else if(!_asyncTaskList.isEmpty())
{
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java Wed Nov 9 14:39:48 2016
@@ -20,7 +20,6 @@
*/
package org.apache.qpid.server.protocol.v0_8;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.qpid.QpidException;
@@ -291,11 +290,6 @@ public abstract class ConsumerTarget_0_8
private final Boolean _autoClose;
- private final AtomicBoolean _deleted = new AtomicBoolean(false);
-
-
-
-
public ConsumerTarget_0_8(AMQChannel channel,
AMQShortString consumerTag,
FieldTable arguments,
@@ -365,16 +359,21 @@ public abstract class ConsumerTarget_0_8
@Override
public boolean isFlowSuspended()
{
- return getState()!=State.ACTIVE || _channel.isSuspended() || _deleted.get() || _channel.getAMQPConnection().isConnectionStopped();
+ // (credit || tcp backpressure) || (channel flow || channel closing || connection closing) || connection being stopped by vh
+ return getState()!=State.ACTIVE || _channel.isSuspended() || _channel.getAMQPConnection().isConnectionStopped();
}
- /**
- * Callback indicating that a queue has been deleted.
- *
- */
- public void queueDeleted()
+ @Override
+ public void updateNotifyWorkDesired()
{
- _deleted.set(true);
+ final AMQPConnection_0_8<?> amqpConnection = _channel.getAMQPConnection();
+
+ boolean state = _channel.isChannelFlow()
+ && !amqpConnection.isTransportBlockedForWriting()
+ // this last condition does not need to exist for browsers and no ack
+ && getCreditManager().hasCredit();
+
+ setNotifyWorkDesired(state);
}
public boolean isAutoClose()
@@ -395,7 +394,13 @@ public abstract class ConsumerTarget_0_8
public boolean allocateCredit(ServerMessage msg)
{
- return _creditManager.useCreditForMessage(msg.getSize());
+ boolean hasCredit = _creditManager.hasCredit();
+ boolean allocated = _creditManager.useCreditForMessage(msg.getSize());
+ if(hasCredit != _creditManager.hasCredit())
+ {
+ _channel.updateAllConsumerNotifyWorkDesired();
+ }
+ return allocated;
}
public AMQChannel getChannel()
@@ -415,7 +420,12 @@ public abstract class ConsumerTarget_0_8
public void restoreCredit(final ServerMessage message)
{
+ boolean hasCredit = _creditManager.hasCredit();
_creditManager.restoreCredit(1, message.getSize());
+ if(_creditManager.hasCredit() != hasCredit)
+ {
+ _channel.updateAllConsumerNotifyWorkDesired();
+ }
}
public void creditStateChanged(boolean hasCredit)
@@ -482,12 +492,10 @@ public abstract class ConsumerTarget_0_8
private void removeUnacknowledgedMessage(MessageInstance entry)
{
-
- final long _size = entry.getMessage().getSize();
- _unacknowledgedBytes.addAndGet(-_size);
+ _unacknowledgedBytes.addAndGet(-entry.getMessage().getSize());
_unacknowledgedCount.decrementAndGet();
- _creditManager.restoreCredit(1, _size);
+ restoreCredit(entry.getMessage());
}
@Override
Copied: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/InfiniteCreditCreditManager.java (from r1768978, qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageOnlyCreditManager.java)
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/InfiniteCreditCreditManager.java?p2=qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/InfiniteCreditCreditManager.java&p1=qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageOnlyCreditManager.java&r1=1768978&r2=1768979&rev=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageOnlyCreditManager.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/InfiniteCreditCreditManager.java Wed Nov 9 14:39:48 2016
@@ -20,54 +20,28 @@
*/
package org.apache.qpid.server.protocol.v0_8;
-import java.util.concurrent.atomic.AtomicLong;
-
import org.apache.qpid.server.flow.AbstractFlowCreditManager;
import org.apache.qpid.server.flow.FlowCreditManager;
-public class MessageOnlyCreditManager extends AbstractFlowCreditManager implements FlowCreditManager
+public class InfiniteCreditCreditManager extends AbstractFlowCreditManager implements FlowCreditManager
{
- private final AtomicLong _messageCredit;
- public MessageOnlyCreditManager(final long initialCredit)
+ public InfiniteCreditCreditManager()
{
- _messageCredit = new AtomicLong(initialCredit);
}
public void restoreCredit(long messageCredit, long bytesCredit)
{
- _messageCredit.addAndGet(messageCredit);
- setSuspended(false);
-
}
public boolean hasCredit()
{
- return _messageCredit.get() > 0L;
+ return true;
}
public boolean useCreditForMessage(long msgSize)
{
- if(hasCredit())
- {
- if(_messageCredit.addAndGet(-1L) >= 0)
- {
- setSuspended(false);
- return true;
- }
- else
- {
- _messageCredit.addAndGet(1L);
- setSuspended(true);
- return false;
- }
- }
- else
- {
- setSuspended(true);
- return false;
- }
-
+ return true;
}
}
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0.java Wed Nov 9 14:39:48 2016
@@ -1405,6 +1405,7 @@ public class AMQPConnection_1_0 extends
public void sendConnectionCloseAsync(final AMQConstant cause, final String message)
{
+ stopConnection();
Action<ConnectionHandler> action = new Action<ConnectionHandler>()
{
@Override
@@ -1552,7 +1553,7 @@ public class AMQPConnection_1_0 extends
@Override
public boolean hasNext()
{
- return !(_sessionsWithWork.isEmpty() && _asyncTaskList.isEmpty());
+ return (!_sessionsWithWork.isEmpty() && !isClosed() && !isConnectionStopped()) || !_asyncTaskList.isEmpty();
}
@Override
@@ -1560,23 +1561,37 @@ public class AMQPConnection_1_0 extends
{
if(!_sessionsWithWork.isEmpty())
{
- if(!_sessionIterator.hasNext())
+ if(isClosed() || isConnectionStopped())
{
- _sessionIterator = _sessionsWithWork.iterator();
+ return new Runnable()
+ {
+ @Override
+ public void run()
+ {
+
+ }
+ };
}
- final AMQSessionModel<?> session = _sessionIterator.next();
- return new Runnable()
+ else
{
- @Override
- public void run()
+ if (!_sessionIterator.hasNext())
{
- _sessionIterator.remove();
- if(session.processPending())
+ _sessionIterator = _sessionsWithWork.iterator();
+ }
+ final AMQSessionModel<?> session = _sessionIterator.next();
+ return new Runnable()
+ {
+ @Override
+ public void run()
{
- _sessionsWithWork.add(session);
+ _sessionIterator.remove();
+ if (session.processPending())
+ {
+ _sessionsWithWork.add(session);
+ }
}
- }
- };
+ };
+ }
}
else if(!_asyncTaskList.isEmpty())
{
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java Wed Nov 9 14:39:48 2016
@@ -33,7 +33,6 @@ import org.apache.qpid.server.message.Me
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.plugin.MessageConverter;
import org.apache.qpid.server.protocol.AMQSessionModel;
-import org.apache.qpid.server.protocol.LinkRegistry;
import org.apache.qpid.server.protocol.MessageConverterRegistry;
import org.apache.qpid.server.protocol.v1_0.codec.ValueHandler;
import org.apache.qpid.server.protocol.v1_0.messaging.SectionEncoder;
@@ -52,6 +51,7 @@ import org.apache.qpid.server.protocol.v
import org.apache.qpid.server.protocol.v1_0.type.transaction.TransactionalState;
import org.apache.qpid.server.protocol.v1_0.type.transport.SenderSettleMode;
import org.apache.qpid.server.protocol.v1_0.type.transport.Transfer;
+import org.apache.qpid.server.transport.AMQPConnection;
import org.apache.qpid.server.transport.ProtocolEngine;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
@@ -88,6 +88,19 @@ class ConsumerTarget_1_0 extends Abstrac
public boolean isFlowSuspended()
{
return _link.getSession().getAMQPConnection().isConnectionStopped() || getState() != State.ACTIVE;
+ }
+
+ @Override
+ public void updateNotifyWorkDesired()
+ {
+ final AMQPConnection<?> amqpConnection =
+ _link.getSession().getAMQPConnection();
+
+ boolean state = !amqpConnection.isTransportBlockedForWriting()
+ && _link.isAttached()
+ && getEndpoint().hasCreditToSend();
+
+ setNotifyWorkDesired(state);
}
@@ -254,6 +267,10 @@ class ConsumerTarget_1_0 extends Abstrac
// TODO
}
+ /*
+ Currently if a queue is deleted the consumer sits there withiout being closed, but
+ obviously not receiving any new messages
+
public void queueDeleted()
{
//TODO
@@ -265,7 +282,7 @@ class ConsumerTarget_1_0 extends Abstrac
.getLinkRegistry(getEndpoint().getSession().getConnection().getRemoteContainerId());
linkReg.unregisterSendingLink(getEndpoint().getName());
}
-
+ */
public boolean allocateCredit(final ServerMessage msg)
{
ProtocolEngine protocolEngine = getSession().getConnection();
@@ -309,6 +326,8 @@ class ConsumerTarget_1_0 extends Abstrac
public void flowStateChanged()
{
+ updateNotifyWorkDesired();
+
ProtocolEngine protocolEngine = getSession().getConnection();
if (isFlowSuspended() && getEndpoint() != null && !protocolEngine.isTransportBlockedForWriting())
{
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java Wed Nov 9 14:39:48 2016
@@ -398,11 +398,6 @@ public class SendingLink_1_0 implements
}
- public void resume(SendingLinkAttachment linkAttachment)
- {
- _linkAttachment = linkAttachment;
- }
-
public void remoteDetached(final LinkEndpoint endpoint, final Detach detach)
{
//TODO
@@ -410,7 +405,7 @@ public class SendingLink_1_0 implements
if(Boolean.TRUE.equals(detach.getClosed())
|| !(TerminusDurability.UNSETTLED_STATE.equals(_durability)|| TerminusDurability.CONFIGURATION.equals(_durability)))
{
- _consumer.close();
+ _target.close();
Modified state = new Modified();
state.setDeliveryFailed(true);
Modified: qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
URL: http://svn.apache.org/viewvc/qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java?rev=1768979&r1=1768978&r2=1768979&view=diff
==============================================================================
--- qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java (original)
+++ qpid/java/branches/remove-queue-runner/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java Wed Nov 9 14:39:48 2016
@@ -30,6 +30,7 @@ import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
@@ -107,6 +108,8 @@ public class Session_1_0 implements AMQS
{
private static final Logger _logger = LoggerFactory.getLogger(Session_1_0.class);
private static final Symbol LIFETIME_POLICY = Symbol.valueOf("lifetime-policy");
+ private static final EnumSet<SessionState> END_STATES =
+ EnumSet.of(SessionState.END_RECVD, SessionState.END_PIPE, SessionState.END_SENT, SessionState.ENDED);
private final AccessControlContext _accessControllerContext;
private final SecurityToken _securityToken;
private AutoCommitTransaction _transaction;
@@ -1530,7 +1533,7 @@ public class Session_1_0 implements AMQS
@Override
public boolean processPending()
{
- if (!getAMQPConnection().isIOThread())
+ if (!getAMQPConnection().isIOThread() || END_STATES.contains(getState()))
{
return false;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org