[1/2] activemq-artemis git commit: This closes #1095

Previous Topic Next Topic
 
classic Classic list List threaded Threaded
2 messages Options
Reply | Threaded
Open this post in threaded view
|

[1/2] activemq-artemis git commit: This closes #1095

clebertsuconic-2
Repository: activemq-artemis
Updated Branches:
  refs/heads/master b674c6b84 -> 9e6c40a8d


This closes #1095


Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/9e6c40a8
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/9e6c40a8
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/9e6c40a8

Branch: refs/heads/master
Commit: 9e6c40a8dec22ac50c1d29b8222a68a75f9a4292
Parents: b674c6b 7282b68
Author: Clebert Suconic <[hidden email]>
Authored: Tue Mar 14 17:13:05 2017 -0400
Committer: Clebert Suconic <[hidden email]>
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(-)
----------------------------------------------------------------------


Reply | Threaded
Open this post in threaded view
|

[2/2] activemq-artemis git commit: ARTEMIS-1039 Transaction Coordinator credit refill

clebertsuconic-2
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 <[hidden email]>
Authored: Tue Mar 14 16:07:58 2017 -0400
Committer: Clebert Suconic <[hidden email]>
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());