Skip to content

Withhold link credit until the connection is started #76

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ansd wants to merge 1 commit into from
Closed

Withhold link credit until the connection is started #76

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
}
Expand Down
3 changes: 3 additions & 0 deletions qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
}
}
Expand Down
2 changes: 2 additions & 0 deletions qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
72 changes: 70 additions & 2 deletions qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand All @@ -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();
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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);
Expand All @@ -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);
Expand All @@ -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();
Expand All @@ -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);
Expand All @@ -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);
Expand All @@ -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();
Expand Down
6 changes: 6 additions & 0 deletions qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -984,6 +984,8 @@ public void testCreateConsumerSourceContainsTempTopicCapabilityWithoutClientID()
private void doCreateConsumerSourceContainsCapabilityTestImpl(Class<? extends Destination> 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);
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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);
Expand Down
8 changes: 8 additions & 0 deletions ...ms-client/src/test/java/org/apache/qpid/jms/integration/SubscriptionsIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions ...s-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down