diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java index 0923f9755..c17b95fc4 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java @@ -141,7 +141,16 @@ public void onPendingFailure(ProviderException cause) { } public void init() throws JMSException { - if (!isPullConsumer()){ + // Link credit is what makes the remote start sending, so it must not be granted while the + // connection is still in stopped mode. A consumer created before Connection.start() gets + // its credit from JmsSession.start() instead. + if (session.isStarted()) { + startConsumerResourceIfNeeded(); + } + } + + void startConsumerResourceIfNeeded() throws JMSException { + if (!isPullConsumer()) { startConsumerResource(); } } diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java index 31d0cd322..572db285d 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java @@ -1179,6 +1179,9 @@ protected void start() throws JMSException { if (started.compareAndSet(false, true)) { for (JmsMessageConsumer consumer : consumers.values()) { consumer.start(); + // Consumers created while the connection was stopped were not granted credit at + // the time, so grant it now that messages may be delivered to them. + consumer.startConsumerResourceIfNeeded(); } } } diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java index 7b54c9b87..22a7b64e8 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java @@ -634,6 +634,8 @@ public void testRemotelyEndConnectionWithSessionWithConsumer() throws Exception try (TestAmqpPeer testPeer = new TestAmqpPeer();) { final Connection connection = testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); testPeer.expectBegin(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java index 659405635..22e02be60 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java @@ -97,11 +97,67 @@ public class ConsumerIntegrationTest extends QpidJmsTestCase { private final IntegrationTestFixture testFixture = new IntegrationTestFixture(); + @Test + @Timeout(20) + public void testConsumerOnStoppedConnectionIsGrantedNoCreditUntilStarted() throws Exception { + try (TestAmqpPeer testPeer = new TestAmqpPeer();) { + // Deliberately left in stopped mode. + Connection connection = testFixture.establishConnecton(testPeer); + testPeer.expectBegin(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue queue = session.createQueue("myQueue"); + + // Attach only. Granting credit here would let the remote send to a connection that has + // not been started, taking messages other consumers on the queue could have had. + testPeer.expectReceiverAttach(); + session.createConsumer(queue); + testPeer.waitForAllHandlersToComplete(1000); + + // Starting the connection is what releases the credit. + testPeer.expectLinkFlow(); + connection.start(); + testPeer.waitForAllHandlersToComplete(1000); + + testPeer.expectClose(); + connection.close(); + + testPeer.waitForAllHandlersToComplete(1000); + } + } + + @Test + @Timeout(20) + public void testConsumerCreatedOnStartedConnectionIsGrantedCreditImmediately() throws Exception { + try (TestAmqpPeer testPeer = new TestAmqpPeer();) { + Connection connection = testFixture.establishConnecton(testPeer); + connection.start(); + testPeer.expectBegin(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue queue = session.createQueue("myQueue"); + + // Connection.start() has already happened and is not repeated, so the credit has to + // come from consumer creation itself. + testPeer.expectReceiverAttach(); + testPeer.expectLinkFlow(); + session.createConsumer(queue); + testPeer.waitForAllHandlersToComplete(1000); + + testPeer.expectClose(); + connection.close(); + + testPeer.waitForAllHandlersToComplete(1000); + } + } + @Test @Timeout(20) public void testCloseConsumer() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { Connection connection = testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); testPeer.expectBegin(); testPeer.expectReceiverAttach(); testPeer.expectLinkFlow(); @@ -125,6 +181,8 @@ public void testCloseConsumer() throws Exception { public void testCloseConsumerTimesOut() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); connection.setCloseTimeout(500); testPeer.expectBegin(); @@ -160,6 +218,8 @@ public void testRemotelyCloseConsumer() throws Exception { CountDownLatch exceptionFired = new CountDownLatch(1); JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); connection.setExceptionListener(new ExceptionListener() { @Override public void onException(JMSException exception) { @@ -628,7 +688,7 @@ public void onMessage(Message m) { @Test @Timeout(20) - public void testNoReceivedMessagesWhenConnectionNotStarted() throws Exception { + public void testNoReceivedMessagesWhenConnectionStopped() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { final CountDownLatch incoming = new CountDownLatch(1); Connection connection = testFixture.establishConnecton(testPeer); @@ -642,6 +702,8 @@ public void onInboundMessage(JmsInboundMessageDispatch envelope) { } }); + // Started, so the consumer is granted credit and the messages reach the client. + connection.start(); testPeer.expectBegin(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); @@ -655,6 +717,8 @@ public void onInboundMessage(JmsInboundMessageDispatch envelope) { // Wait for a message to arrive then try and receive it, which should not happen // since the connection is not started. assertTrue(incoming.await(10, TimeUnit.SECONDS)); + // Pausing delivery must hide them from the application again. + connection.stop(); assertNull(consumer.receive(100)); testPeer.expectClose(); @@ -666,7 +730,7 @@ public void onInboundMessage(JmsInboundMessageDispatch envelope) { @Test @Timeout(20) - public void testNoReceivedNoWaitMessagesWhenConnectionNotStarted() throws Exception { + public void testNoReceivedNoWaitMessagesWhenConnectionStopped() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { final CountDownLatch incoming = new CountDownLatch(1); Connection connection = testFixture.establishConnecton(testPeer); @@ -680,6 +744,8 @@ public void onInboundMessage(JmsInboundMessageDispatch envelope) { } }); + // Started, so the consumer is granted credit and the messages reach the client. + connection.start(); testPeer.expectBegin(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); @@ -693,6 +759,8 @@ public void onInboundMessage(JmsInboundMessageDispatch envelope) { // Wait for a message to arrive then try and receive it, which should not happen // since the connection is not started. assertTrue(incoming.await(10, TimeUnit.SECONDS)); + // Pausing delivery must hide them from the application again. + connection.stop(); assertNull(consumer.receiveNoWait()); testPeer.expectClose(); diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java index c4c06d41d..08ff68512 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java @@ -984,6 +984,8 @@ public void testCreateConsumerSourceContainsTempTopicCapabilityWithoutClientID() private void doCreateConsumerSourceContainsCapabilityTestImpl(Class destType, boolean setClientID) throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { Connection connection = testFixture.establishConnecton(testPeer, false, null, null, null, setClientID); + // Consumers only get link credit once the connection is started. + connection.start(); testPeer.expectBegin(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); @@ -2000,6 +2002,8 @@ public void testRemotelyEndSessionWithConsumers() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { final CountDownLatch sessionClosed = new CountDownLatch(1); JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); connection.addConnectionListener(new JmsDefaultConnectionListener() { @Override public void onSessionClosed(Session session, Throwable exception) { @@ -2090,6 +2094,8 @@ public boolean isSatisfied() throws Exception { public void testCloseSessionWithConsumerThatRemoteDetaches() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { Connection connection = testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); testPeer.expectBegin(); Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SubscriptionsIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SubscriptionsIntegrationTest.java index 5786ac38f..98548be67 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SubscriptionsIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SubscriptionsIntegrationTest.java @@ -1191,6 +1191,8 @@ public void testRemotelyDetachLinkWithDurableSharedConsumer() throws Exception { // Establish connection Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); + // Consumers only get link credit once the connection is started. + connection.start(); final CountDownLatch subscriberClosed = new CountDownLatch(1); ((JmsConnection) connection).addConnectionListener(new JmsDefaultConnectionListener() { @@ -1452,6 +1454,8 @@ public void testRemotelyEndSessionWithDurableSharedConsumer() throws Exception { // Establish connection Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); + // Consumers only get link credit once the connection is started. + connection.start(); final CountDownLatch sessionClosed = new CountDownLatch(1); ((JmsConnection) connection).addConnectionListener(new JmsDefaultConnectionListener() { @@ -1537,6 +1541,8 @@ public void testLocallyEndSessionWithSharedConsumer() throws Exception { // Establish connection Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); + // Consumers only get link credit once the connection is started. + connection.start(); final CountDownLatch sessionClosed = new CountDownLatch(1); ((JmsConnection) connection).addConnectionListener(new JmsDefaultConnectionListener() { @@ -1694,6 +1700,8 @@ public void testSharedTopicSubscriberBehavesLikeNoClientIDWasSetWhenAwaitClientI ConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:" + serverPort + "?jms.awaitClientID=false"); Connection connection = factory.createConnection(); + // Consumers only get link credit once the connection is started. + connection.start(); // Verify that all handlers complete, i.e. the awaitClientID=false option // setting was effective in provoking the AMQP Open immediately even diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java index 6e176d4cb..d09270de9 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java @@ -2914,6 +2914,8 @@ private void doRemotelyCloseConsumerWithMessageListenerFiresJMSExceptionListener testPeer.expectBegin(); final JmsConnection connection = establishAnonymousConnecton("failover.maxReconnectAttempts=1", testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); connection.setExceptionListener(new ExceptionListener() { @Override public void onException(JMSException exception) {