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/02/19 16:24:28 UTC
svn commit: r1660909 - in /qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java:
broker-core/src/main/java/org/apache/qpid/server/connection/
broker-core/src/main/java/org/apache/qpid/server/model/
broker-core/src/main/java/org/apache/qpid/server/model/adap...
Author: kwall
Date: Thu Feb 19 15:24:27 2015
New Revision: 1660909
URL: http://svn.apache.org/r1660909
Log:
Connection close is now performed by i/o thread
Added:
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/CloseFuture.java
- copied, changed from r1659606, qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/ServerProtocolEngine.java
Modified:
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/AbstractConfiguredObject.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConnectionAdapter.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/ServerProtocolEngine.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/actors/BaseConnectionActorTestCase.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/model/VirtualHostTest.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/systests/src/test/java/org/apache/qpid/client/AsynchMessageListenerTest.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/systests/src/test/java/org/apache/qpid/systest/management/jmx/ConnectionManagementTest.java
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java Thu Feb 19 15:24:27 2015
@@ -74,7 +74,7 @@ public class ConnectionRegistry implemen
AMQConnectionModel connection = itr.next();
try
{
- connection.close(AMQConstant.CONNECTION_FORCED, replyText);
+ connection.closeAsync(AMQConstant.CONNECTION_FORCED, replyText);
}
catch (Exception e)
{
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/AbstractConfiguredObject.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/AbstractConfiguredObject.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/AbstractConfiguredObject.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/AbstractConfiguredObject.java Thu Feb 19 15:24:27 2015
@@ -490,16 +490,37 @@ public abstract class AbstractConfigured
{
if(_dynamicState.compareAndSet(DynamicState.OPENED, DynamicState.CLOSED))
{
- beforeClose();
- closeChildren();
- onClose();
- unregister(false);
+ CloseFuture close = beforeClose();
+
+ Runnable closeRunnable = new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ closeChildren();
+ onClose();
+ unregister(false);
+
+ }
+ };
+
+ if (close == null)
+ {
+ closeRunnable.run();
+ }
+ else
+ {
+ close.runWhenComplete(closeRunnable);
+ }
+
+ // if future not complete, schedule the remainder to be done once complete.
}
}
- protected void beforeClose()
+ protected CloseFuture beforeClose()
{
+ return null;
}
protected void onClose()
Copied: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/CloseFuture.java (from r1659606, qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/ServerProtocolEngine.java)
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/CloseFuture.java?p2=qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/CloseFuture.java&p1=qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/ServerProtocolEngine.java&r1=1659606&r2=1660909&rev=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/ServerProtocolEngine.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/CloseFuture.java Thu Feb 19 15:24:27 2015
@@ -1,5 +1,4 @@
/*
- *
* 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
@@ -16,39 +15,12 @@
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
- *
*/
-package org.apache.qpid.server.protocol;
-import javax.security.auth.Subject;
+package org.apache.qpid.server.model;
-import org.apache.qpid.protocol.ProtocolEngine;
-import org.apache.qpid.server.util.Action;
-public interface ServerProtocolEngine extends ProtocolEngine
+public interface CloseFuture
{
- /**
- * Gets the connection ID associated with this ProtocolEngine
- */
- long getConnectionId();
-
- Subject getSubject();
-
- boolean isTransportBlockedForWriting();
-
- void setTransportBlockedForWriting(boolean blocked);
-
- void setMessageAssignmentSuspended(boolean value);
-
- boolean isMessageAssignmentSuspended();
-
- void processPendingMessages();
-
- boolean hasWork();
-
- void clearWork();
-
- void notifyWork();
-
- void setWorkListener(Action<ServerProtocolEngine> listener);
+ public void runWhenComplete(final Runnable closeRunnable);
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConnectionAdapter.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConnectionAdapter.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConnectionAdapter.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConnectionAdapter.java Thu Feb 19 15:24:27 2015
@@ -31,6 +31,7 @@ import org.apache.log4j.Logger;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.model.AbstractConfiguredObject;
+import org.apache.qpid.server.model.CloseFuture;
import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.Connection;
import org.apache.qpid.server.model.Port;
@@ -51,6 +52,7 @@ public final class ConnectionAdapter ext
private final Action _underlyingConnectionDeleteTask;
private final AtomicBoolean _underlyingClosed = new AtomicBoolean(false);
private AMQConnectionModel _underlyingConnection;
+ private final AtomicBoolean _closing = new AtomicBoolean();
public ConnectionAdapter(final AMQConnectionModel conn)
{
@@ -158,15 +160,42 @@ public final class ConnectionAdapter ext
@StateTransition( currentState = State.ACTIVE, desiredState = State.DELETED)
private void doDelete()
{
- closeUnderlyingConnection();
+ asyncClose();
deleted();
setState(State.DELETED);
}
@Override
+ protected CloseFuture beforeClose()
+ {
+ _closing.set(true);
+
+ final ConnectionCloseFuture closeFuture = asyncClose();
+
+ return closeFuture;
+ }
+
+ private ConnectionCloseFuture asyncClose()
+ {
+ final ConnectionCloseFuture closeFuture = new ConnectionCloseFuture();
+
+ _underlyingConnection.addDeleteTask(new Action()
+ {
+ @Override
+ public void performAction(final Object object)
+ {
+ LOGGER.debug("KWDEBUG underlying connection deleted");
+ closeFuture.connectionClosed();
+ }
+ });
+
+ _underlyingConnection.closeAsync(AMQConstant.CONNECTION_FORCED, "Connection closed by external action");
+ return closeFuture;
+ }
+
+ @Override
protected void onClose()
{
- closeUnderlyingConnection();
}
@Override
@@ -233,23 +262,54 @@ public final class ConnectionAdapter ext
// SessionAdapter installs delete task to cause session model object to delete
}
- private void closeUnderlyingConnection()
+
+ private static class ConnectionCloseFuture implements CloseFuture
{
- if (_underlyingClosed.compareAndSet(false, true))
+ private boolean _closed;
+
+ public synchronized void connectionClosed()
+ {
+ _closed = true;
+ notifyAll();
+
+ }
+
+ @Override
+ public void runWhenComplete(final Runnable closeRunnable)
{
- _underlyingConnection.removeDeleteTask(_underlyingConnectionDeleteTask);
- try
+ if (_closed )
{
- _underlyingConnection.close(AMQConstant.CONNECTION_FORCED, "Connection closed by external action");
+ closeRunnable.run();
}
- catch (Exception e)
+ else
{
- LOGGER.warn("Exception closing connection "
- + _underlyingConnection.getConnectionId()
- + " from "
- + _underlyingConnection.getRemoteAddressString(), e);
- }
+ Thread t = new Thread(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ synchronized (ConnectionCloseFuture.this)
+ {
+ while (!_closed)
+ {
+ try
+ {
+ ConnectionCloseFuture.this.wait();
+ }
+ catch (InterruptedException e)
+ {
+ }
+ }
+
+ closeRunnable.run();
+ }
+ }
+ });
+
+ t.setDaemon(true);
+ t.start();
+ }
}
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java Thu Feb 19 15:24:27 2015
@@ -40,7 +40,7 @@ public interface AMQConnectionModel<T ex
* @param cause
* @param message
*/
- public void close(AMQConstant cause, String message);
+ public void closeAsync(AMQConstant cause, String message);
public void block();
@@ -110,4 +110,5 @@ public interface AMQConnectionModel<T ex
void notifyWork();
boolean isMessageAssignmentSuspended();
+
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java Thu Feb 19 15:24:27 2015
@@ -214,9 +214,9 @@ public class MultiVersionProtocolEngine
}
@Override
- public void processPendingMessages()
+ public void processPending()
{
- _delegate.processPendingMessages();
+ _delegate.processPending();
}
@Override
@@ -260,7 +260,7 @@ public class MultiVersionProtocolEngine
}
@Override
- public void processPendingMessages()
+ public void processPending()
{
}
@@ -418,7 +418,7 @@ public class MultiVersionProtocolEngine
}
@Override
- public void processPendingMessages()
+ public void processPending()
{
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/ServerProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/ServerProtocolEngine.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/ServerProtocolEngine.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/ServerProtocolEngine.java Thu Feb 19 15:24:27 2015
@@ -42,7 +42,7 @@ public interface ServerProtocolEngine ex
boolean isMessageAssignmentSuspended();
- void processPendingMessages();
+ void processPending();
boolean hasWork();
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java Thu Feb 19 15:24:27 2015
@@ -756,10 +756,10 @@ public abstract class AbstractQueue<X ex
}
@Override
- protected void beforeClose()
+ protected org.apache.qpid.server.model.CloseFuture beforeClose()
{
_closing = true;
- super.beforeClose();
+ return super.beforeClose();
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java Thu Feb 19 15:24:27 2015
@@ -287,7 +287,7 @@ public class NonBlockingConnection imple
_protocolEngine.setMessageAssignmentSuspended(true);
- _protocolEngine.processPendingMessages();
+ _protocolEngine.processPending();
_protocolEngine.setTransportBlockedForWriting(!doWrite());
boolean dataRead = doRead();
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java Thu Feb 19 15:24:27 2015
@@ -677,9 +677,10 @@ public abstract class AbstractVirtualHos
}
@Override
- protected void beforeClose()
+ protected CloseFuture beforeClose()
{
setState(State.UNAVAILABLE);
+ return null;
}
@Override
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java Thu Feb 19 15:24:27 2015
@@ -538,7 +538,7 @@ public class MockConsumer implements Con
}
@Override
- public void close(AMQConstant cause, String message)
+ public void closeAsync(AMQConstant cause, String message)
{
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/actors/BaseConnectionActorTestCase.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/actors/BaseConnectionActorTestCase.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/actors/BaseConnectionActorTestCase.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/actors/BaseConnectionActorTestCase.java Thu Feb 19 15:24:27 2015
@@ -55,7 +55,7 @@ public abstract class BaseConnectionActo
}
if (_session != null)
{
- _session.close(AMQConstant.CONNECTION_FORCED, "");
+ _session.closeAsync(AMQConstant.CONNECTION_FORCED, "");
}
}
finally
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/model/VirtualHostTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/model/VirtualHostTest.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/model/VirtualHostTest.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-core/src/test/java/org/apache/qpid/server/model/VirtualHostTest.java Thu Feb 19 15:24:27 2015
@@ -251,7 +251,7 @@ public class VirtualHostTest extends Qpi
0,
virtualHost.getChildren(Connection.class).size());
- verify(connection).close(AMQConstant.CONNECTION_FORCED, "Connection closed by external action");
+ verify(connection).closeAsync(AMQConstant.CONNECTION_FORCED, "Connection closed by external action");
}
public void testDeleteVirtualHost_ClosesConnections()
@@ -276,7 +276,7 @@ public class VirtualHostTest extends Qpi
0,
virtualHost.getChildren(Connection.class).size());
- verify(connection).close(AMQConstant.CONNECTION_FORCED, "Connection closed by external action");
+ verify(connection).closeAsync(AMQConstant.CONNECTION_FORCED, "Connection closed by external action");
}
public void testCreateDurableQueue()
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java Thu Feb 19 15:24:27 2015
@@ -291,12 +291,10 @@ public class ProtocolEngine_0_10 extend
}
@Override
- public void processPendingMessages()
+ public void processPending()
{
- for (AMQSessionModel session : _connection.getSessionModels())
- {
- session.processPendingMessages();
- }
+ _connection.processPending();
+
}
@Override
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java Thu Feb 19 15:24:27 2015
@@ -23,6 +23,7 @@ package org.apache.qpid.server.protocol.
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CONNECTION_FORMAT;
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SOCKET_FORMAT;
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.USER_FORMAT;
+import static org.apache.qpid.transport.Connection.State.CLOSING;
import java.net.SocketAddress;
import java.security.Principal;
@@ -30,6 +31,8 @@ import java.security.PrivilegedAction;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -81,9 +84,12 @@ public class ServerConnection extends Co
private boolean _blocking;
private Transport _transport;
- private final CopyOnWriteArrayList<Action<? super ServerConnection>> _taskList =
+ private final CopyOnWriteArrayList<Action<? super ServerConnection>> _connectionCloseTaskList =
new CopyOnWriteArrayList<Action<? super ServerConnection>>();
+ private final Queue<Action<? super ServerConnection>> _asyncTaskList =
+ new ConcurrentLinkedQueue<>();
+
private final CopyOnWriteArrayList<SessionModelListener> _sessionListeners =
new CopyOnWriteArrayList<SessionModelListener>();
@@ -368,25 +374,35 @@ public class ServerConnection extends Co
}
}
- public void close(AMQConstant cause, String message)
+ public void closeAsync(final AMQConstant cause, final String message)
{
- closeSubscriptions();
- performDeleteTasks();
- ConnectionCloseCode replyCode = ConnectionCloseCode.NORMAL;
- try
- {
- replyCode = ConnectionCloseCode.get(cause.getCode());
- }
- catch (IllegalArgumentException iae)
+
+ addAsyncTask(new Action<ServerConnection>()
{
- // Ignore
- }
- close(replyCode, message);
+ @Override
+ public void performAction(final ServerConnection object)
+ {
+ closeSubscriptions();
+ performDeleteTasks();
+
+ setState(CLOSING);
+ ConnectionCloseCode replyCode = ConnectionCloseCode.NORMAL;
+ try
+ {
+ replyCode = ConnectionCloseCode.get(cause.getCode());
+ }
+ catch (IllegalArgumentException iae)
+ {
+ // Ignore
+ }
+ sendConnectionClose(replyCode, message);
+ }
+ });
}
protected void performDeleteTasks()
{
- for(Action<? super ServerConnection> task : _taskList)
+ for(Action<? super ServerConnection> task : _connectionCloseTaskList)
{
task.performAction(this);
}
@@ -659,13 +675,19 @@ public class ServerConnection extends Co
@Override
public void addDeleteTask(final Action<? super ServerConnection> task)
{
- _taskList.add(task);
+ _connectionCloseTaskList.add(task);
+ }
+
+ private void addAsyncTask(final Action<ServerConnection> action)
+ {
+ _asyncTaskList.add(action);
+ notifyWork();
}
@Override
public void removeDeleteTask(final Action<? super ServerConnection> task)
{
- _taskList.remove(task);
+ _connectionCloseTaskList.remove(task);
}
public int getMessageCompressionThreshold()
@@ -698,4 +720,19 @@ public class ServerConnection extends Co
{
return _serverProtocolEngine.isMessageAssignmentSuspended();
}
+
+ public void processPending()
+ {
+ while(_asyncTaskList.peek() != null)
+ {
+ Action<? super ServerConnection> asyncAction = _asyncTaskList.poll();
+ asyncAction.performAction(this);
+ }
+
+ for (AMQSessionModel session : getSessionModels())
+ {
+ session.processPendingMessages();
+ }
+
+ }
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java Thu Feb 19 15:24:27 2015
@@ -439,7 +439,7 @@ public class ServerSessionDelegate exten
}
catch (VirtualHostUnavailableException e)
{
- getServerConnection(serverSession).close(AMQConstant.CONNECTION_FORCED, e.getMessage());
+ getServerConnection(serverSession).closeAsync(AMQConstant.CONNECTION_FORCED, e.getMessage());
}
finally
{
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java Thu Feb 19 15:24:27 2015
@@ -1714,7 +1714,7 @@ public class AMQChannel
receivedLock.lock();
try
{
- _connection.close(AMQConstant.RESOURCE_ERROR, reason);
+ _connection.closeAsync(AMQConstant.RESOURCE_ERROR, reason);
}
finally
{
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java Thu Feb 19 15:24:27 2015
@@ -36,8 +36,10 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -164,9 +166,12 @@ public class AMQProtocolEngine implement
/* AMQP Version for this session */
private ProtocolVersion _protocolVersion = ProtocolVersion.getLatestSupportedVersion();
private final MethodRegistry _methodRegistry = new MethodRegistry(_protocolVersion);
- private final List<Action<? super AMQProtocolEngine>> _taskList =
+ private final List<Action<? super AMQProtocolEngine>> _connectionCloseTaskList =
new CopyOnWriteArrayList<>();
+ private final Queue<Action<? super AMQProtocolEngine>> _asyncTaskList =
+ new ConcurrentLinkedQueue<>();
+
private Map<Integer, Long> _closingChannelsList = new ConcurrentHashMap<>();
private ProtocolOutputConverter _protocolOutputConverter;
private final Subject _authorizedSubject = new Subject();
@@ -849,6 +854,7 @@ public class AMQProtocolEngine implement
finally
{
_receivedLock.unlock();
+
finishClose(connectionDropped);
}
@@ -890,7 +896,7 @@ public class AMQProtocolEngine implement
{
try
{
- for (Action<? super AMQProtocolEngine> task : _taskList)
+ for (Action<? super AMQProtocolEngine> task : _connectionCloseTaskList)
{
task.performAction(this);
}
@@ -975,13 +981,15 @@ public class AMQProtocolEngine implement
try
{
markChannelAwaitingCloseOk(channelId);
- closeSession(false);
+ closeSession(false); // currently performs the delete actions.
}
finally
{
try
{
writeFrame(frame);
+
+ // add an async job and not
}
finally
{
@@ -1133,12 +1141,12 @@ public class AMQProtocolEngine implement
public void addDeleteTask(Action<? super AMQProtocolEngine> task)
{
- _taskList.add(task);
+ _connectionCloseTaskList.add(task);
}
public void removeDeleteTask(Action<? super AMQProtocolEngine> task)
{
- _taskList.remove(task);
+ _connectionCloseTaskList.remove(task);
}
public ProtocolOutputConverter getProtocolOutputConverter()
@@ -1388,11 +1396,29 @@ public class AMQProtocolEngine implement
writeFrame(responseBody.generateFrame(channelId));
}
- public void close(AMQConstant cause, String message)
+ public void closeAsync(final AMQConstant cause, final String message)
{
- closeConnection(0, new AMQConnectionException(cause, message, 0, 0,
- getMethodRegistry(),
- null));
+ _logger.debug("KWDEBUG About to schedule close");
+
+ Action<AMQProtocolEngine> action = new Action<AMQProtocolEngine>()
+ {
+ @Override
+ public void performAction(final AMQProtocolEngine object)
+ {
+ _logger.debug("KWDEBUG About to perform close");
+ closeConnection(0, new AMQConnectionException(cause, message, 0, 0,
+ getMethodRegistry(),
+ null));
+
+ }
+ };
+ addAsyncTask(action);
+ }
+
+ private void addAsyncTask(final Action<AMQProtocolEngine> action)
+ {
+ _asyncTaskList.add(action);
+ notifyWork();
}
public void block()
@@ -2080,8 +2106,14 @@ public class AMQProtocolEngine implement
}
@Override
- public void processPendingMessages()
+ public void processPending()
{
+ while(_asyncTaskList.peek() != null)
+ {
+ Action<? super AMQProtocolEngine> asyncAction = _asyncTaskList.poll();
+ asyncAction.performAction(this);
+ }
+
for (AMQSessionModel session : getSessionModels())
{
session.processPendingMessages();
@@ -2100,7 +2132,6 @@ public class AMQProtocolEngine implement
_stateChanged.set(true);
final Action<ServerProtocolEngine> listener = _workListener.get();
- _logger.info("Work lister is null? " + (listener == null));
if(listener != null)
{
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java Thu Feb 19 15:24:27 2015
@@ -30,11 +30,14 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import javax.security.auth.Subject;
+import org.apache.qpid.AMQConnectionException;
import org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint;
import org.apache.qpid.amqp_1_0.transport.ConnectionEventListener;
import org.apache.qpid.amqp_1_0.transport.LinkEndpoint;
@@ -58,6 +61,7 @@ import org.apache.qpid.server.security.a
import org.apache.qpid.server.stats.StatisticsCounter;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.virtualhost.VirtualHostImpl;
+import org.apache.qpid.transport.Connection;
public class Connection_1_0 implements ConnectionEventListener, AMQConnectionModel<Connection_1_0,Session_1_0>
{
@@ -100,6 +104,11 @@ public class Connection_1_0 implements C
private List<Action<? super Connection_1_0>> _closeTasks =
Collections.synchronizedList(new ArrayList<Action<? super Connection_1_0>>());
+
+ private final Queue<Action<? super Connection_1_0>> _asyncTaskList =
+ new ConcurrentLinkedQueue<>();
+
+
private boolean _closedOnOpen;
@@ -213,6 +222,13 @@ public class Connection_1_0 implements C
_closeTasks.add( task );
}
+ private void addAsyncTask(final Action<Connection_1_0> action)
+ {
+ _asyncTaskList.add(action);
+ notifyWork();
+ }
+
+
public void closeReceived()
{
Collection<Session_1_0> sessions = new ArrayList(_sessions);
@@ -251,9 +267,19 @@ public class Connection_1_0 implements C
@Override
- public void close(AMQConstant cause, String message)
+ public void closeAsync(AMQConstant cause, String message)
{
- _conn.close();
+ Action<Connection_1_0> action = new Action<Connection_1_0>()
+ {
+ @Override
+ public void performAction(final Connection_1_0 object)
+ {
+ _conn.close();
+
+ }
+ };
+ addAsyncTask(action);
+
}
@Override
@@ -510,4 +536,19 @@ public class Connection_1_0 implements C
{
return _protocolEngine.isMessageAssignmentSuspended();
}
+
+ public void processPending()
+ {
+ while(_asyncTaskList.peek() != null)
+ {
+ Action<? super Connection_1_0> asyncAction = _asyncTaskList.poll();
+ asyncAction.performAction(this);
+ }
+
+ for (AMQSessionModel session : getSessionModels())
+ {
+ session.processPendingMessages();
+ }
+
+ }
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java Thu Feb 19 15:24:27 2015
@@ -620,12 +620,10 @@ public class ProtocolEngine_1_0_0_SASL i
}
@Override
- public void processPendingMessages()
+ public void processPending()
{
- for (AMQSessionModel session : _connection.getSessionModels())
- {
- session.processPendingMessages();
- }
+ _connection.processPending();
+
}
@Override
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java Thu Feb 19 15:24:27 2015
@@ -616,6 +616,12 @@ public class Connection extends Connecti
close(ConnectionCloseCode.CONNECTION_FORCED, "The connection was closed using the broker's management interface.");
}
+
+ protected void sendConnectionClose(ConnectionCloseCode replyCode, String replyText, Option ... _options)
+ {
+ connectionClose(replyCode, replyText, _options);
+ }
+
public void close(ConnectionCloseCode replyCode, String replyText, Option ... _options)
{
synchronized (lock)
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/systests/src/test/java/org/apache/qpid/client/AsynchMessageListenerTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/systests/src/test/java/org/apache/qpid/client/AsynchMessageListenerTest.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/systests/src/test/java/org/apache/qpid/client/AsynchMessageListenerTest.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/systests/src/test/java/org/apache/qpid/client/AsynchMessageListenerTest.java Thu Feb 19 15:24:27 2015
@@ -70,6 +70,9 @@ public class AsynchMessageListenerTest e
}
+
+
+
public void testMessageListener() throws Exception
{
CountingMessageListener countingMessageListener = new CountingMessageListener(MSG_COUNT);
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/systests/src/test/java/org/apache/qpid/systest/management/jmx/ConnectionManagementTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/systests/src/test/java/org/apache/qpid/systest/management/jmx/ConnectionManagementTest.java?rev=1660909&r1=1660908&r2=1660909&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/systests/src/test/java/org/apache/qpid/systest/management/jmx/ConnectionManagementTest.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/systests/src/test/java/org/apache/qpid/systest/management/jmx/ConnectionManagementTest.java Thu Feb 19 15:24:27 2015
@@ -23,13 +23,7 @@ import java.util.Date;
import java.util.Iterator;
import java.util.List;
-import javax.jms.Connection;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.Queue;
-import javax.jms.Session;
+import javax.jms.*;
import javax.management.JMException;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.CompositeDataSupport;
@@ -74,6 +68,38 @@ public class ConnectionManagementTest ex
}
}
+ public void testManagementClosesConnection() throws Exception
+ {
+ assertEquals("Expected no managed connections", 0, getManagedConnections().size());
+
+ _connection = getConnection();
+ assertEquals("Expected one managed connection", 1, getManagedConnections().size());
+
+
+ ManagedConnection managedConnection = getManagedConnections().get(0);
+
+ managedConnection.closeConnection();
+
+ assertEquals("Expected no managed connections", 0, getManagedConnections().size());
+
+ /*
+ try
+ {
+
+ _connection.start();
+ fail("Exception not thrown");
+ }
+ catch (javax.jms.IllegalStateException ise)
+ {
+ ise.printStackTrace();
+ // PASS
+ }*/
+
+ }
+
+
+
+
public void testNumberOfManagedConnectionsMatchesNumberOfClientConnections() throws Exception
{
assertEquals("Expected no managed connections", 0, getManagedConnections().size());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org