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

ARTEMIS-4108 Improving flush on large message fix

This commit is contained in:
Clebert Suconic 2022-12-06 16:07:09 -05:00
parent eb11b044af
commit 33567fca8d
1 changed files with 21 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
View File

@ -191,13 +191,22 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
// If the draining is already running, then don't do anything // If the draining is already running, then don't do anything
if (draining.compareAndSet(false, true)) { if (draining.compareAndSet(false, true)) {
final ProtonServerSenderContext plugSender = (ProtonServerSenderContext) serverConsumer.getProtocolContext(); final ProtonServerSenderContext plugSender = (ProtonServerSenderContext) serverConsumer.getProtocolContext();
flushDrain(serverConsumer, plugSender);
}
} else {
serverConsumer.receiveCredits(-1);
}
}
private void flushDrain(ServerConsumerImpl serverConsumer, ProtonServerSenderContext plugSender) {
serverConsumer.forceDelivery(1, new Runnable() { serverConsumer.forceDelivery(1, new Runnable() {
@Override @Override
public void run() { public void run() {
try { try {
connection.runNow(() -> { connection.runNow(() -> {
if (pendingLargeMessage != null) { if (pendingLargeMessage != null) {
afterLargeMessage = () -> drained(plugSender); // retry the flush after the large message is done
afterLargeMessage = () -> flushDrain(serverConsumer, plugSender);
} else { } else {
drained(plugSender); drained(plugSender);
} }
@ -208,10 +217,6 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
} }
}); });
} }
} else {
serverConsumer.receiveCredits(-1);
}
}
private void drained(ProtonServerSenderContext sender) { private void drained(ProtonServerSenderContext sender) {
sender.reportDrained(); sender.reportDrained();