Add sync/async varients to most ACO methods

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/QPID-6262-JavaBrokerNIO@1665306 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 52 insertions, 15 deletions

diff --git a/qpid/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/plugins/ACLFileAccessControlProviderImpl.java b/qpid/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/plugins/ACLFileAccessControlProviderImpl.java
index 00b42094b1..c1bd1b0bb8 100644
--- a/qpid/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/plugins/ACLFileAccessControlProviderImpl.java
+++ b/qpid/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/plugins/ACLFileAccessControlProviderImpl.java
@@ -26,6 +26,9 @@ import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
 
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
 import org.apache.log4j.Logger;
 
 import org.apache.qpid.server.configuration.IllegalConfigurationException;
@@ -152,7 +155,7 @@ public class ACLFileAccessControlProviderImpl
 
 
     @StateTransition(currentState = {State.UNINITIALIZED, State.QUIESCED, State.ERRORED}, desiredState = State.ACTIVE)
-    private void activate()
+    private ListenableFuture<Void> activate()
     {
 
         if(_broker.isManagementMode())
@@ -180,6 +183,7 @@ public class ACLFileAccessControlProviderImpl
                 }
             }
         }
+        return Futures.immediateFuture(null);
     }
 
     @Override
@@ -193,17 +197,36 @@ public class ACLFileAccessControlProviderImpl
     }
 
     @StateTransition(currentState = State.UNINITIALIZED, desiredState = State.QUIESCED)
-    private void startQuiesced()
+    private ListenableFuture<Void> startQuiesced()
     {
         setState(State.QUIESCED);
+        return Futures.immediateFuture(null);
     }
 
     @StateTransition(currentState = {State.ACTIVE, State.QUIESCED, State.ERRORED}, desiredState = State.DELETED)
-    private void doDelete()
+    private ListenableFuture<Void> doDelete()
     {
-        close();
-        setState(State.DELETED);
-        deleted();
+        final SettableFuture<Void> returnVal = SettableFuture.create();
+        closeAsync().addListener(
+                new Runnable()
+                {
+                    @Override
+                    public void run()
+                    {
+                        try
+                        {
+
+                            setState(State.DELETED);
+                            deleted();
+                        }
+                        finally
+                        {
+                            returnVal.set(null);
+                        }
+                    }
+                }, getTaskExecutor().getExecutor()
+                           );
+        return returnVal;
     }
 
     @Override
diff --git a/qpid/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/ACLFileAccessControlProviderFactoryTest.java b/qpid/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/ACLFileAccessControlProviderFactoryTest.java
index a34ac16e80..2a691b3652 100644
--- a/qpid/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/ACLFileAccessControlProviderFactoryTest.java
+++ b/qpid/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/ACLFileAccessControlProviderFactoryTest.java
@@ -30,6 +30,7 @@ import java.util.UUID;
 import java.util.regex.Pattern;
 
 import org.apache.qpid.server.configuration.IllegalConfigurationException;
+import org.apache.qpid.server.configuration.updater.CurrentThreadTaskExecutor;
 import org.apache.qpid.server.configuration.updater.TaskExecutor;
 import org.apache.qpid.server.model.AccessControlProvider;
 import org.apache.qpid.server.model.Broker;
@@ -54,7 +55,9 @@ public class ACLFileAccessControlProviderFactoryTest extends QpidTestCase
         when(_broker.getObjectFactory()).thenReturn(_objectFactory);
         when(_broker.getModel()).thenReturn(_objectFactory.getModel());
         when(_broker.getCategoryClass()).thenReturn(Broker.class);
-        when(_broker.getTaskExecutor()).thenReturn(mock(TaskExecutor.class));
+        TaskExecutor taskExecutor = new CurrentThreadTaskExecutor();
+        taskExecutor.start();
+        when(_broker.getTaskExecutor()).thenReturn(taskExecutor);
     }
 
     public void testCreateInstanceWhenAclFileIsNotPresent()
diff --git a/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java b/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
index 62a95e9869..28d8a6c88c 100644
--- a/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
+++ b/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
@@ -125,9 +125,8 @@ class ManagementNodeConsumer implements ConsumerImpl
     }
 
     @Override
-    public ListenableFuture<Void> close()
+    public void close()
     {
-        return Futures.immediateFuture(null);
     }
 
     @Override
diff --git a/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/HttpManagement.java b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/HttpManagement.java
index 69920ff488..1a85a24e0b 100644
--- a/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/HttpManagement.java
+++ b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/HttpManagement.java
@@ -39,6 +39,8 @@ import javax.servlet.DispatcherType;
 import javax.servlet.MultipartConfigElement;
 import javax.servlet.http.HttpServletRequest;
 
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 import org.apache.log4j.Logger;
 import org.eclipse.jetty.server.Connector;
 import org.eclipse.jetty.server.Server;
@@ -130,7 +132,7 @@ public class HttpManagement extends AbstractPluginAdapter<HttpManagement> implem
     }
 
     @StateTransition(currentState = {State.UNINITIALIZED,State.ERRORED}, desiredState = State.ACTIVE)
-    private void doStart()
+    private ListenableFuture<Void> doStart()
     {
         getBroker().getEventLogger().message(ManagementConsoleMessages.STARTUP(OPERATIONAL_LOGGING_NAME));
 
@@ -148,6 +150,7 @@ public class HttpManagement extends AbstractPluginAdapter<HttpManagement> implem
 
         getBroker().getEventLogger().message(ManagementConsoleMessages.READY(OPERATIONAL_LOGGING_NAME));
         setState(State.ACTIVE);
+        return Futures.immediateFuture(null);
     }
 
     @Override
@@ -206,7 +209,9 @@ public class HttpManagement extends AbstractPluginAdapter<HttpManagement> implem
 
                 if(port.getState() != State.ACTIVE)
                 {
-                    port.start();
+
+                    // TODO - RG
+                    port.startAsync();
                 }
                 Connector connector = null;
 
diff --git a/qpid/java/broker-plugins/management-http/src/test/java/org/apache/qpid/server/management/plugin/HttpManagementTest.java b/qpid/java/broker-plugins/management-http/src/test/java/org/apache/qpid/server/management/plugin/HttpManagementTest.java
index 52d7ba33a3..4327292336 100644
--- a/qpid/java/broker-plugins/management-http/src/test/java/org/apache/qpid/server/management/plugin/HttpManagementTest.java
+++ b/qpid/java/broker-plugins/management-http/src/test/java/org/apache/qpid/server/management/plugin/HttpManagementTest.java
@@ -30,6 +30,7 @@ import java.util.Map;
 import java.util.UUID;
 
 import org.apache.qpid.server.configuration.updater.TaskExecutor;
+import org.apache.qpid.server.configuration.updater.TaskExecutorImpl;
 import org.apache.qpid.server.logging.EventLogger;
 import org.apache.qpid.server.model.AuthenticationProvider;
 import org.apache.qpid.server.model.Broker;
@@ -58,7 +59,9 @@ public class HttpManagementTest extends QpidTestCase
         when(_broker.getModel()).thenReturn(objectFactory.getModel());
         when(_broker.getCategoryClass()).thenReturn(Broker.class);
         when(_broker.getEventLogger()).thenReturn(mock(EventLogger.class));
-        when(_broker.getTaskExecutor()).thenReturn(mock(TaskExecutor.class));
+        TaskExecutor taskExecutor = new TaskExecutorImpl();
+        taskExecutor.start();
+        when(_broker.getTaskExecutor()).thenReturn(taskExecutor);
 
         Map<String, Object> attributes = new HashMap<String, Object>();
         attributes.put(HttpManagement.HTTP_BASIC_AUTHENTICATION_ENABLED, false);
diff --git a/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/JMXManagementPluginImpl.java b/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/JMXManagementPluginImpl.java
index 6c962c2901..06558b9f9a 100644
--- a/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/JMXManagementPluginImpl.java
+++ b/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/JMXManagementPluginImpl.java
@@ -32,6 +32,8 @@ import java.util.Set;
 import javax.management.InstanceAlreadyExistsException;
 import javax.management.JMException;
 
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 import org.apache.log4j.Logger;
 
 import org.apache.qpid.server.configuration.IllegalConfigurationException;
@@ -105,7 +107,7 @@ public class JMXManagementPluginImpl
     }
 
     @StateTransition(currentState = {State.UNINITIALIZED,State.ERRORED}, desiredState = State.ACTIVE)
-    private void doStart() throws JMException, IOException
+    private ListenableFuture<Void> doStart() throws JMException, IOException
     {
         _allowPortActivation = true;
         Broker<?> broker = getBroker();
@@ -125,7 +127,8 @@ public class JMXManagementPluginImpl
                 registryPort.setPortManager(this);
                 if(port.getState() != State.ACTIVE)
                 {
-                    port.start();
+                    // TODO - RG
+                    port.startAsync();
                 }
 
             }
@@ -135,7 +138,7 @@ public class JMXManagementPluginImpl
                 connectorPort.setPortManager(this);
                 if(port.getState() != State.ACTIVE)
                 {
-                    port.start();
+                    port.startAsync();
                 }
 
             }
@@ -175,6 +178,7 @@ public class JMXManagementPluginImpl
         _objectRegistry.start();
         setState(State.ACTIVE);
         _allowPortActivation = false;
+        return Futures.immediateFuture(null);
     }
 
     @Override