Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

This closes #3029

This commit is contained in:
Clebert Suconic 2020-04-13 16:57:06 -04:00
parent 30d521152d fc6402613d
commit 89634e3660
1 changed files with 8 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQConsumer.java
View File

@ -68,7 +68,7 @@ public class AMQConsumer {
private int prefetchSize;
private final AtomicInteger currentWindow;
private final AtomicInteger deliveredAcks;
private int deliveredAcks;
private long messagePullSequence = 0;
private final AtomicReference<MessagePullHandler> messagePullHandler = new AtomicReference<>(null);
//internal means we don't expose
@ -88,7 +88,7 @@ public class AMQConsumer {
this.scheduledPool = scheduledPool;
this.prefetchSize = info.getPrefetchSize();
this.currentWindow = new AtomicInteger(prefetchSize);
this.deliveredAcks = new AtomicInteger(0);
this.deliveredAcks = 0;
if (prefetchSize == 0) {
messagePullHandler.set(new MessagePullHandler());
}
@ -300,18 +300,15 @@ public class AMQConsumer {
List<MessageReference> ackList = serverConsumer.getDeliveringReferencesBasedOnProtocol(removeReferences, first, last);
if (removeReferences && (ack.isIndividualAck() || ack.isStandardAck() || ack.isPoisonAck())) {
this.deliveredAcks.getAndUpdate(deliveredAcks -> {
if (deliveredAcks >= ackList.size()) {
return deliveredAcks - ackList.size();
}
if (deliveredAcks < ackList.size()) {
acquireCredit(ackList.size() - deliveredAcks);
return 0;
});
deliveredAcks = 0;
} else {
deliveredAcks -= ackList.size();
}
} else {
if (ack.isDeliveredAck()) {
this.deliveredAcks.addAndGet(ack.getMessageCount());
this.deliveredAcks += ack.getMessageCount();
}
acquireCredit(ack.getMessageCount());