ARTEMIS-1039 Transaction Coordinator credit refill The coordinator needs to refill credit on the receiver once it has been exhausted, otherwise the remote cannot send additional declare or discharge commands to the broker.
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/7282b689 Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/7282b689 Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/7282b689 Branch: refs/heads/master Commit: 7282b6890aaafce7209daa89311715586f9b4564 Parents: b674c6b Author: Timothy Bish <[email protected]> Authored: Tue Mar 14 16:07:58 2017 -0400 Committer: Clebert Suconic <[email protected]> Committed: Tue Mar 14 17:13:05 2017 -0400 ---------------------------------------------------------------------- .../protocol/amqp/proton/AMQPSessionContext.java | 2 +- .../transaction/ProtonTransactionHandler.java | 13 +++++++++++-- .../tests/integration/amqp/AmqpTransactionTest.java | 16 ++++++++++++++++ 3 files changed, 28 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7282b689/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java ---------------------------------------------------------------------- diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java index 89b6ed3..d1fc0e1 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java @@ -148,7 +148,7 @@ public class AMQPSessionContext extends ProtonInitializable { receiver.setContext(transactionHandler); receiver.open(); - receiver.flow(100); + receiver.flow(ProtonTransactionHandler.DEFAULT_COORDINATOR_CREDIT); } public void addSender(Sender sender) throws Exception { http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7282b689/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionHandler.java ---------------------------------------------------------------------- diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionHandler.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionHandler.java index 2cdb072..721bd33 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionHandler.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionHandler.java @@ -16,8 +16,6 @@ */ package org.apache.activemq.artemis.protocol.amqp.proton.transaction; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.PooledByteBufAllocator; import org.apache.activemq.artemis.protocol.amqp.broker.AMQPSessionCallback; import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPException; import org.apache.activemq.artemis.protocol.amqp.logger.ActiveMQAMQPProtocolMessageBundle; @@ -38,6 +36,9 @@ import org.apache.qpid.proton.engine.Receiver; import org.apache.qpid.proton.message.impl.MessageImpl; import org.jboss.logging.Logger; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.PooledByteBufAllocator; + /** * handles an amqp Coordinator to deal with transaction boundaries etc */ @@ -45,6 +46,8 @@ public class ProtonTransactionHandler implements ProtonDeliveryHandler { private static final Logger log = Logger.getLogger(ProtonTransactionHandler.class); + public static final int DEFAULT_COORDINATOR_CREDIT = 100; + final AMQPSessionCallback sessionSPI; public ProtonTransactionHandler(AMQPSessionCallback sessionSPI) { @@ -98,6 +101,12 @@ public class ProtonTransactionHandler implements ProtonDeliveryHandler { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCommittingCoordinator(e.getMessage()); } } + + // Replenish coordinator receiver credit on exhaustion so sender can continue + // transaction declare and discahrge operations. + if (receiver.getCredit() == 0) { + receiver.flow(DEFAULT_COORDINATOR_CREDIT); + } } } catch (ActiveMQAMQPException amqpE) { delivery.disposition(createRejected(amqpE.getAmqpError(), amqpE.getMessage())); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7282b689/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpTransactionTest.java ---------------------------------------------------------------------- diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpTransactionTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpTransactionTest.java index 1b2a1b0..c00cc1c 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpTransactionTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpTransactionTest.java @@ -49,6 +49,22 @@ public class AmqpTransactionTest extends AmqpClientTestSupport { } @Test(timeout = 30000) + public void testCoordinatorReplenishesCredit() throws Exception { + AmqpClient client = createAmqpClient(); + AmqpConnection connection = addConnection(client.connect()); + AmqpSession session = connection.createSession(); + assertNotNull(session); + + for (int i = 0; i < 1000; ++i) { + session.begin(); + assertTrue(session.isInTransaction()); + session.commit(); + } + + connection.close(); + } + + @Test(timeout = 30000) public void testBeginAndRollbackTransaction() throws Exception { AmqpClient client = createAmqpClient(); AmqpConnection connection = addConnection(client.connect());
