Apache/activemq-artemis
1
0
Fork 0
mirror of https://github.com/apache/activemq-artemis.git synced 2025-03-06 01:10:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-1604 Artemis deadlock using MQTT Protocol

Browse Source 
Address code review comment not address when PR was merged.
This commit is contained in:
Michael André Pearce 2019-03-12 19:53:07 +00:00
parent ef7d9800be
commit fafbd7e2e5
1 changed files with 51 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
View File

@ -3094,60 +3094,64 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
return false;
}
try {
synchronized (this) {
if (!supportsDirectDeliver) {
return false;
}
if (paused || !canDispatch() && redistributor == null) {
return false;
return deliver(ref);
} finally {
deliverLock.unlock();
}
}
private boolean deliver(final MessageReference ref) {
synchronized (this) {
if (!supportsDirectDeliver) {
return false;
}
if (paused || !canDispatch() && redistributor == null) {
return false;
}
if (checkExpired(ref)) {
return true;
}
consumers.reset();
while (consumers.hasNext() || redistributor != null) {
ConsumerHolder<? extends Consumer> holder = redistributor == null ? consumers.next() : redistributor;
Consumer consumer = holder.consumer;
final SimpleString groupID = extractGroupID(ref);
Consumer groupConsumer = getGroupConsumer(groupID);
if (groupConsumer != null) {
consumer = groupConsumer;
}
if (checkExpired(ref)) {
HandleStatus status = handle(ref, consumer);
if (status == HandleStatus.HANDLED) {
if (redistributor == null) {
handleMessageGroup(ref, consumer, groupConsumer, groupID);
}
messagesAdded.incrementAndGet();
deliveriesInTransit.countUp();
proceedDeliver(consumer, ref);
consumers.reset();
return true;
}
consumers.reset();
while (consumers.hasNext() || redistributor != null) {
ConsumerHolder<? extends Consumer> holder = redistributor == null ? consumers.next() : redistributor;
Consumer consumer = holder.consumer;
final SimpleString groupID = extractGroupID(ref);
Consumer groupConsumer = getGroupConsumer(groupID);
if (groupConsumer != null) {
consumer = groupConsumer;
}
HandleStatus status = handle(ref, consumer);
if (status == HandleStatus.HANDLED) {
if (redistributor == null) {
handleMessageGroup(ref, consumer, groupConsumer, groupID);
}
messagesAdded.incrementAndGet();
deliveriesInTransit.countUp();
proceedDeliver(consumer, ref);
consumers.reset();
return true;
}
if (redistributor != null || groupConsumer != null) {
break;
}
if (redistributor != null || groupConsumer != null) {
break;
}
if (logger.isTraceEnabled()) {
logger.tracef("Queue " + getName() + " is out of direct delivery as no consumers handled a delivery");
}
return false;
}
} finally {
deliverLock.unlock();
if (logger.isTraceEnabled()) {
logger.tracef("Queue " + getName() + " is out of direct delivery as no consumers handled a delivery");
}
return false;
}
}