You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2015/10/29 23:58:49 UTC
svn commit: r1711375 - in /qpid/java/trunk:
client/src/main/java/org/apache/qpid/client/
systests/src/test/java/org/apache/qpid/test/unit/client/connection/
Author: kwall
Date: Thu Oct 29 22:58:48 2015
New Revision: 1711375
URL: http://svn.apache.org/viewvc?rev=1711375&view=rev
Log:
QPID-6780: [Java Client 0-10] Ensure that sessions, producers, consumers and associated threads are closed when connection is closed by peer or dropped
Modified:
qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession.java
qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/client/connection/BrokerClosesClientConnectionTest.java
Modified: qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java?rev=1711375&r1=1711374&r2=1711375&view=diff
==============================================================================
--- qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java (original)
+++ qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java Thu Oct 29 22:58:48 2015
@@ -365,6 +365,10 @@ public class AMQConnectionDelegate_0_10
}
+ for(AMQSession<?,?> session : _conn.getSessions().values())
+ {
+ session.markClosed();
+ }
_conn.setClosed();
Modified: qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession.java?rev=1711375&r1=1711374&r2=1711375&view=diff
==============================================================================
--- qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession.java Thu Oct 29 22:58:48 2015
@@ -2225,9 +2225,7 @@ public abstract class AMQSession<C exten
/**
* Called to mark the session as being closed. Useful when the session needs to be made invalid, e.g. after failover
- * when the client has veoted resubscription.
- * <p>
- * The caller of this method must already hold the failover mutex.
+ * when the client has vetoed resubscription.
*/
void markClosed()
{
Modified: qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/client/connection/BrokerClosesClientConnectionTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/client/connection/BrokerClosesClientConnectionTest.java?rev=1711375&r1=1711374&r2=1711375&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/client/connection/BrokerClosesClientConnectionTest.java (original)
+++ qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/client/connection/BrokerClosesClientConnectionTest.java Thu Oct 29 22:58:48 2015
@@ -27,10 +27,13 @@ import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
-import javax.naming.NamingException;
+import javax.jms.Queue;
import org.apache.qpid.AMQConnectionClosedException;
import org.apache.qpid.AMQDisconnectedException;
import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.BasicMessageConsumer;
+import org.apache.qpid.client.BasicMessageProducer;
import org.apache.qpid.test.utils.QpidBrokerTestCase;
import org.apache.qpid.transport.ConnectionException;
@@ -49,6 +52,8 @@ public class BrokerClosesClientConnectio
private boolean _isExternalBroker;
private final RecordingExceptionListener _recordingExceptionListener = new RecordingExceptionListener();
private Session _session;
+ private MessageConsumer _consumer;
+ private MessageProducer _producer;
@Override
protected void setUp() throws Exception
@@ -56,8 +61,12 @@ public class BrokerClosesClientConnectio
super.setUp();
_connection = getConnection();
+ _connection.start();
_session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
_connection.setExceptionListener(_recordingExceptionListener);
+ Queue queue = _session.createTemporaryQueue();
+ _consumer = _session.createConsumer(queue);
+ _producer = _session.createProducer(queue);
_isExternalBroker = isExternalBroker();
}
@@ -66,13 +75,13 @@ public class BrokerClosesClientConnectio
{
final Class<? extends Exception> expectedLinkedException = isBroker010() ? ConnectionException.class : AMQConnectionClosedException.class;
- assertConnectionOpen();
+ assertJmsObjectsOpen();
stopBroker();
JMSException exception = _recordingExceptionListener.awaitException(10000);
assertConnectionCloseWasReported(exception, expectedLinkedException);
- assertConnectionClosed();
+ assertJmsObjectsClosed();
ensureCanCloseWithoutException();
}
@@ -86,13 +95,13 @@ public class BrokerClosesClientConnectio
return;
}
- assertConnectionOpen();
+ assertJmsObjectsOpen();
killBroker();
JMSException exception = _recordingExceptionListener.awaitException(10000);
assertConnectionCloseWasReported(exception, expectedLinkedException);
- assertConnectionClosed();
+ assertJmsObjectsClosed();
ensureCanCloseWithoutException();
}
@@ -117,14 +126,20 @@ public class BrokerClosesClientConnectio
assertEquals("Unexpected linked exception", linkedExceptionClass, exception.getLinkedException().getClass());
}
- private void assertConnectionClosed()
+ private void assertJmsObjectsClosed()
{
- assertTrue("Connection should be marked as closed", ((AMQConnection)_connection).isClosed());
+ assertTrue("Connection should be marked as closed", ((AMQConnection) _connection).isClosed());
+ assertTrue("Session should be marked as closed", ((AMQSession) _session).isClosed());
+ assertTrue("Producer should be marked as closed", ((BasicMessageProducer) _producer).isClosed());
+ assertTrue("Consumer should be marked as closed", ((BasicMessageConsumer) _consumer).isClosed());
}
- private void assertConnectionOpen()
+ private void assertJmsObjectsOpen()
{
- assertFalse("Connection should not be marked as closed", ((AMQConnection)_connection).isClosed());
+ assertFalse("Connection should not be marked as closed", ((AMQConnection) _connection).isClosed());
+ assertFalse("Session should not be marked as closed", ((AMQSession) _session).isClosed());
+ assertFalse("Producer should not be marked as closed", ((BasicMessageProducer) _producer).isClosed());
+ assertFalse("Consumer should not be marked as closed", ((BasicMessageConsumer) _consumer).isClosed());
}
private final class RecordingExceptionListener implements ExceptionListener
@@ -162,7 +177,7 @@ public class BrokerClosesClientConnectio
}
}
- public void testNoDeliveryAfterBrokerClose() throws JMSException, NamingException, InterruptedException
+ public void testNoDeliveryAfterBrokerClose() throws Exception
{
Listener listener = new Listener();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org