From 478a46b504e3a6577af388ea737e4dafcc24334a Mon Sep 17 00:00:00 2001 From: Martin Ritchie Date: Fri, 10 Apr 2009 23:46:19 +0000 Subject: QPID-1779 : Application of patches attached to JIRA. Should address connection close issues experienced on 0-8/9 branch Excluded test from TCP runs as it is hardwired to InVM. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@764109 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/java/010ExcludeList | 3 + qpid/java/08ExcludeList-nonvm | 2 + .../java/org/apache/qpid/client/AMQConnection.java | 10 +- .../java/org/apache/qpid/client/AMQSession.java | 22 +++-- .../apache/qpid/client/BasicMessageConsumer.java | 5 +- .../java/org/apache/qpid/client/Closeable.java | 18 ++++ .../CloseAfterConnectionFailureTest.java | 110 +++++++++++++++++++++ 7 files changed, 158 insertions(+), 12 deletions(-) create mode 100644 qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/CloseAfterConnectionFailureTest.java (limited to 'qpid/java') diff --git a/qpid/java/010ExcludeList b/qpid/java/010ExcludeList index 94102ce09c..936dd762d5 100644 --- a/qpid/java/010ExcludeList +++ b/qpid/java/010ExcludeList @@ -64,3 +64,6 @@ org.apache.qpid.test.client.QueueBrowsingFlowToDiskTest#* // This test currently does not pick up the runtime location of the nonVm queueBacking store. org.apache.qpid.test.unit.client.close.FlowToDiskBackingQueueDeleteTest#* +// This test may use QpidTestCase but it is not using the getConnection and is hardwired to InVM +org.apache.qpid.test.unit.client.connection.CloseAfterConnectionFailureTest#* + diff --git a/qpid/java/08ExcludeList-nonvm b/qpid/java/08ExcludeList-nonvm index a529a2fa87..c2f263c870 100644 --- a/qpid/java/08ExcludeList-nonvm +++ b/qpid/java/08ExcludeList-nonvm @@ -32,3 +32,5 @@ org.apache.qpid.client.MessageListenerTest#testSynchronousRecieveNoWait // This test currently does not pick up the runtime location of the nonVm queueBacking store. org.apache.qpid.test.unit.client.close.FlowToDiskBackingQueueDeleteTest#* +// This test may use QpidTestCase but it is not using the getConnection and is hardwired to InVM +org.apache.qpid.test.unit.client.connection.CloseAfterConnectionFailureTest#* diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java index 5c48d73e43..c09b05bda8 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java @@ -58,6 +58,7 @@ import org.apache.qpid.AMQConnectionFailureException; import org.apache.qpid.AMQException; import org.apache.qpid.AMQProtocolException; import org.apache.qpid.AMQUnresolvedAddressException; +import org.apache.qpid.AMQDisconnectedException; import org.apache.qpid.client.configuration.ClientProperties; import org.apache.qpid.client.failover.FailoverException; import org.apache.qpid.client.failover.FailoverProtectedOperation; @@ -924,7 +925,12 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect { if (!_closed.getAndSet(true)) { - doClose(sessions, timeout); + _closing.set(true); + try{ + doClose(sessions, timeout); + }finally{ + _closing.set(false); + } } } @@ -1318,7 +1324,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect // in the case of an IOException, MINA has closed the protocol session so we set _closed to true // so that any generic client code that tries to close the connection will not mess up this error // handling sequence - if (cause instanceof IOException) + if (cause instanceof IOException || cause instanceof AMQDisconnectedException) { closer = !_closed.getAndSet(true); diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java index b632c56708..2782505191 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java @@ -282,7 +282,7 @@ public abstract class AMQSession extends Closeable implements Messa { try { - sendCancel(); + if (!_connection.isClosing()) + { + sendCancel(); + } } catch (AMQException e) { diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/Closeable.java b/qpid/java/client/src/main/java/org/apache/qpid/client/Closeable.java index 7e119343a1..e6771e122c 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/Closeable.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/Closeable.java @@ -51,6 +51,13 @@ public abstract class Closeable */ protected final AtomicBoolean _closed = new AtomicBoolean(false); + /** + * Are we in the process of closing. We have this distinction so we can + * still signal we are in the process of closing so other objects can tell + * the difference and tidy up. + */ + protected final AtomicBoolean _closing = new AtomicBoolean(false); + /** * Checks if this is closed, and raises a JMSException if it is. * @@ -74,6 +81,17 @@ public abstract class Closeable return _closed.get(); } + /** + * Checks if this is closis. + * + * @return true if we are closing, false otherwise. + */ + public boolean isClosing() + { + return _closing.get(); + } + + /** * Closes this object. * diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/CloseAfterConnectionFailureTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/CloseAfterConnectionFailureTest.java new file mode 100644 index 0000000000..1cb24919f0 --- /dev/null +++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/CloseAfterConnectionFailureTest.java @@ -0,0 +1,110 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.test.unit.client.connection; + +import org.apache.qpid.AMQException; +import org.apache.qpid.client.AMQConnection; +import org.apache.qpid.client.AMQConnectionURL; +import org.apache.qpid.client.AMQSession; +import org.apache.qpid.client.transport.TransportConnection; +import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException; +import org.apache.qpid.test.utils.QpidTestCase; +import org.apache.qpid.url.URLSyntaxException; + +import javax.jms.ExceptionListener; +import javax.jms.JMSException; +import javax.jms.MessageConsumer; +import javax.jms.Session; +import java.util.concurrent.CountDownLatch; + +public class CloseAfterConnectionFailureTest extends QpidTestCase implements ExceptionListener +{ + private int sessionCount = 0; + AMQConnection connection; + Session session; + MessageConsumer consumer; + private CountDownLatch _latch = new CountDownLatch(1); + + public void testNoFailover() throws URLSyntaxException, AMQVMBrokerCreationException, + InterruptedException, JMSException + { + String connectionString = "amqp://guest:guest@/test?brokerlist='vm://:1?connectdelay='500',retries='3'',failover='nofailover'"; + + AMQConnectionURL url = new AMQConnectionURL(connectionString); + + try + { + //Start the connection so it will use the retries + connection = new AMQConnection(url, null); + + connection.setExceptionListener(this); + + session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + consumer = session.createConsumer(session.createQueue(this.getName())); + + //Kill connection + TransportConnection.killAllVMBrokers(); + _latch.await(); + } + catch (AMQException e) + { + fail(e.getMessage()); + } + } + + public void onException(JMSException e) + { + System.err.println("Connection isClosed after connection Falure?:" + connection.isClosed()); + try + { + consumer.close(); + } + catch (JMSException jsme) + { + System.err.println("Consumer close failed with:" + jsme.getMessage()); + } + System.err.println("Connection isClosed after connection Falure?:" + connection.isClosed()); + try + { + //Note that if we actually do session.close() we will lock up as the session will never receive a frame + // from the + ((AMQSession)session).close(10); + } + catch (JMSException jsme) + { + System.err.println("Session close failed with:" + jsme.getMessage()); + } + System.err.println("Connection isClosed after connection Falure?:" + connection.isClosed()); + + try + { + connection.close(); + } + catch (JMSException jsme) + { + System.err.println("Session close failed with:" + jsme.getMessage()); + } + System.err.println("Connection isClosed after connection Falure?:" + connection.isClosed()); + + _latch.countDown(); + } + +} -- cgit v1.2.1