From 1209b7fc315cd3c84bbf959b7ffeac5eb01fa41c Mon Sep 17 00:00:00 2001
From: Robert Gemmell <robbie@apache.org>
Date: Fri, 9 Apr 2010 14:16:08 +0000
Subject: QPID-2379: add Connection.close() method implementation

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@932428 13f79535-47bb-0310-9956-ffa450edef68
---
 .../main/java/org/apache/qpid/qmf/QMFService.java  |  5 ++-
 .../server/configuration/ConnectionConfig.java     |  2 +
 .../apache/qpid/server/federation/BrokerLink.java  |  5 +++
 .../qpid/server/protocol/AMQProtocolEngine.java    | 50 +++++++++++++++++++++-
 .../qpid/server/protocol/ProtocolEngine_0_10.java  |  5 +++
 .../java/org/apache/qpid/transport/Connection.java | 16 +++++--
 6 files changed, 76 insertions(+), 7 deletions(-)

(limited to 'qpid/java')

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFService.java b/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFService.java
index 381c376f56..c32fcfd73a 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFService.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFService.java
@@ -1214,8 +1214,9 @@ public class QMFService implements ConfigStore.ConfigEventListener
 
         public BrokerSchema.ConnectionClass.CloseMethodResponseCommand close(final BrokerSchema.ConnectionClass.CloseMethodResponseCommandFactory factory)
         {
-            //todo
-            throw new UnsupportedOperationException();
+            _obj.mgmtClose();
+            
+            return factory.createResponseCommand();
         }
 
         public UUID getId()
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConnectionConfig.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConnectionConfig.java
index ad451f44a7..0dd36fe1fe 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConnectionConfig.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConnectionConfig.java
@@ -44,4 +44,6 @@ public interface ConnectionConfig extends ConfiguredObject<ConnectionConfigType,
     ConfigStore getConfigStore();
     
     Boolean isShadow();
+    
+    void mgmtClose();
 }
\ No newline at end of file
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java
index ba305c96fa..fa2fb9ead1 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java
@@ -188,6 +188,11 @@ public class BrokerLink implements LinkConfig, ConnectionListener
         {
             return false;
         }
+
+        public void mgmtClose()
+        {
+            _connectionConfig.mgmtClose();
+        }
     }
 
     private class SessionFactory implements Connection.SessionFactory
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
index e2735d151b..01ea7e5bb1 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
@@ -46,6 +46,7 @@ import org.apache.qpid.server.logging.LogActor;
 import org.apache.qpid.server.logging.LogSubject;
 import org.apache.qpid.server.logging.actors.AMQPConnectionActor;
 import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.ManagementActor;
 import org.apache.qpid.server.logging.messages.ConnectionMessages;
 import org.apache.qpid.server.logging.subjects.ConnectionLogSubject;
 import org.apache.qpid.server.management.Managable;
@@ -61,6 +62,7 @@ import org.apache.qpid.transport.NetworkDriver;
 import org.apache.qpid.transport.Sender;
 
 import javax.management.JMException;
+import javax.management.MBeanException;
 import javax.security.sasl.SaslServer;
 import java.io.IOException;
 import java.net.InetSocketAddress;
@@ -703,6 +705,9 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol
                 }
 
                 closeAllChannels();
+                
+                getConfigStore().removeConfiguredObject(this);
+
                 if (_managedObject != null)
                 {
                     _managedObject.unregister();
@@ -763,8 +768,6 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol
 
     public void closeProtocolSession()
     {
-        getConfigStore().removeConfiguredObject(this);
-
         _networkDriver.close();
         try
         {
@@ -1143,5 +1146,48 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol
     {
         return false;
     }
+    
+    public void mgmtClose()
+    {
+        MethodRegistry methodRegistry = getMethodRegistry();
+        ConnectionCloseBody responseBody =
+                methodRegistry.createConnectionCloseBody(
+                        AMQConstant.REPLY_SUCCESS.getCode(),
+                        new AMQShortString("The connection was closed using the broker's management interface."),
+                        0,0);
+
+        // This seems ugly but because we use closeConnection in both normal
+        // broker operation and as part of the management interface it cannot
+        // be avoided. The Current Actor will be null when this method is
+        // called via the QMF management interface. As such we need to set one.
+        boolean removeActor = false;
+        if (CurrentActor.get() == null)
+        {
+            removeActor = true;
+            CurrentActor.set(new ManagementActor(_actor.getRootMessageLogger()));
+        }
+
+        try
+        {
+            writeFrame(responseBody.generateFrame(0));
+
+            try
+            {
+
+                closeSession();
+            }
+            catch (AMQException ex)
+            {
+                throw new RuntimeException(ex);
+            }
+        }
+        finally
+        {
+            if (removeActor)
+            {
+                CurrentActor.remove();
+            }
+        }
+    }
 
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
index 56784e7251..f1e79839c9 100755
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
@@ -188,4 +188,9 @@ public class ProtocolEngine_0_10  extends InputHandler implements ProtocolEngine
     {
         return false;
     }
+    
+    public void mgmtClose()
+    {
+        _connection.mgmtClose();
+    }
 }
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
index 2ca5d28f42..1f559f690f 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
@@ -531,6 +531,16 @@ public class Connection extends ConnectionInvoker
     }
 
     public void close()
+    {
+        close(ConnectionCloseCode.NORMAL, null);
+    }
+    
+    public void mgmtClose()
+    {
+        close(ConnectionCloseCode.CONNECTION_FORCED, "The connection was closed using the broker's management interface.");
+    }
+    
+    public void close(ConnectionCloseCode replyCode, String replyText, Option ... _options)
     {
         synchronized (lock)
         {
@@ -538,7 +548,7 @@ public class Connection extends ConnectionInvoker
             {
             case OPEN:
                 state = CLOSING;
-                connectionClose(ConnectionCloseCode.NORMAL, null);
+                connectionClose(replyCode, replyText, _options);
                 Waiter w = new Waiter(lock, timeout);
                 while (w.hasTime() && state == CLOSING && error == null)
                 {
@@ -547,14 +557,14 @@ public class Connection extends ConnectionInvoker
 
                 if (error != null)
                 {
-                    close();
+                    close(replyCode, replyText, _options);
                     throw new ConnectionException(error);
                 }
 
                 switch (state)
                 {
                 case CLOSING:
-                    close();
+                    close(replyCode, replyText, _options);
                     throw new ConnectionException("close() timed out");
                 case CLOSED:
                     break;
-- 
cgit v1.2.1