Author: rgodfrey
Date: Tue Nov 10 11:09:02 2015
New Revision: 1713610
URL: http://svn.apache.org/viewvc?rev=1713610&view=rev
Log:
QPID-6820 : State change listeners on non-acquired messages do not get removed
when consumers are closed
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java?rev=1713610&r1=1713609&r2=1713610&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
Tue Nov 10 11:09:02 2015
@@ -2215,7 +2215,8 @@ public abstract class AbstractQueue<X ex
//QPID-1187 - Treat the consumer as suspended for this
message
// and wait for the message to be removed to continue
delivery.
subActive = false;
- node.addStateChangeListener(new
QueueEntryListener(sub));
+ sub.awaitCredit(node);
+
}
}
@@ -2625,34 +2626,6 @@ public abstract class AbstractQueue<X ex
}
}
- private final class QueueEntryListener implements
StateChangeListener<MessageInstance, QueueEntry.State>
- {
-
- private final QueueConsumer<?> _sub;
-
- public QueueEntryListener(final QueueConsumer<?> sub)
- {
- _sub = sub;
- }
-
- public boolean equals(Object o)
- {
- return o instanceof AbstractQueue.QueueEntryListener
- && _sub == ((QueueEntryListener) o)._sub;
- }
-
- public int hashCode()
- {
- return System.identityHashCode(_sub);
- }
-
- public void stateChanged(MessageInstance entry, QueueEntry.State
oldSate, QueueEntry.State newState)
- {
- entry.removeStateChangeListener(this);
- deliverAsync();
- }
- }
-
public List<Long> getMessagesOnTheQueue(int num)
{
return getMessagesOnTheQueue(num, 0);
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java?rev=1713610&r1=1713609&r2=1713610&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java
Tue Nov 10 11:09:02 2015
@@ -51,4 +51,5 @@ public interface QueueConsumer<X extends
QueueContext getQueueContext();
+ void awaitCredit(QueueEntry entry);
}
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java?rev=1713610&r1=1713609&r2=1713610&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
Tue Nov 10 11:09:02 2015
@@ -31,6 +31,7 @@ import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,6 +71,7 @@ class QueueConsumerImpl
private final long _consumerNumber;
private final long _createTime = System.currentTimeMillis();
private final MessageInstance.ConsumerAcquiredState<QueueConsumerImpl>
_owningState = new
MessageInstance.ConsumerAcquiredState<QueueConsumerImpl>(this);
+ private final WaitingOnCreditMessageListener
_waitingOnCreditMessageListener = new WaitingOnCreditMessageListener();
private final boolean _acquires;
private final boolean _seesRequeues;
private final boolean _isTransient;
@@ -235,6 +237,12 @@ class QueueConsumerImpl
}
@Override
+ public void awaitCredit(final QueueEntry node)
+ {
+ _waitingOnCreditMessageListener.update(node);
+ }
+
+ @Override
public void externalStateChange()
{
_queue.deliverAsync();
@@ -278,6 +286,7 @@ class QueueConsumerImpl
_target.getSendLock();
try
{
+ _waitingOnCreditMessageListener.remove();
_target.consumerRemoved(this);
_target.removeStateChangeListener(_listener);
_queue.unregisterConsumer(this);
@@ -617,4 +626,44 @@ class QueueConsumerImpl
{
return _queue.getEventLogger();
}
+
+ public class WaitingOnCreditMessageListener implements
StateChangeListener<MessageInstance, QueueEntry.State>
+ {
+ private final AtomicReference<MessageInstance> _entry = new
AtomicReference<>();
+
+ public WaitingOnCreditMessageListener()
+ {
+ }
+
+ public void update(final MessageInstance entry)
+ {
+ remove();
+ // this only happens under send lock so only one thread can be
setting to a non null value at any time
+ _entry.set(entry);
+ entry.addStateChangeListener(this);
+ if(!entry.isAvailable())
+ {
+ _queue.deliverAsync();
+ remove();
+ }
+ }
+
+ public void remove()
+ {
+ MessageInstance instance;
+ if((instance = _entry.getAndSet(null)) != null)
+ {
+ instance.removeStateChangeListener(this);
+ }
+
+ }
+
+ public void stateChanged(MessageInstance entry, QueueEntry.State
oldSate, QueueEntry.State newState)
+ {
+ entry.removeStateChangeListener(this);
+ _entry.compareAndSet(entry, null);
+ _queue.deliverAsync();
+ }
+
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
