Rename shutdown to closeAsync and add close/Closeable

JAVA-247 #fixes

Author: Sylvain Lebresne

driver-core/CHANGELOG.rst

@@ -5,6 +5,8 @@ CHANGELOG
 ------
 
 - [improvement] Make most main objects interface to facilitate testing/mocking (JAVA-195)
+- [api] Renamed shutdown to closeAsync and ShutdownFuture to CloseFuture. Clustering
+  and Session also now implement Closeable (JAVA-247).
 - [bug] Fix potential thread leaks when shutting down Metrics (JAVA-232)
 - [bug] Fix potential NPE in HostConnectionPool (JAVA-231)
 - [bug] Avoid NPE when node is in an unconfigured DC (JAVA-244)

driver-core/Upgrade_guide_to_2.0.rst

@@ -48,10 +48,11 @@ Main API changes:
    in the same JVM). As a result, tools that were polling JMX informations will
    have to be updated accordingly.
 
-5. The Cluster and Session shutdown API has changed. There is now only one
-   shutdown method that is asynchronous but return a Future on the completion
-   of shutdown. Also, shutdown now waits for ongoing queries to complete by
-   default (but you can force the closing of all connections if you want to).
+5. The Cluster and Session shutdown API has changed. There is now a closeAsync
+   that is asynchronous but return a Future on the completion of the shutdown
+   process. And there is also a close shortcut that does the same but block.
+   Also, close now waits for ongoing queries to complete by default (but you
+   can force the closing of all connections if you want to).
 
 6. The QueryBuilder#raw method does not automatically add quotes anymore, but
    rather ouptut its result without an change (as the raw name implies). This

…datastax/driver/core/ShutdownFuture.java …om/datastax/driver/core/CloseFuture.javadriver-core/src/main/java/com/datastax/driver/core/ShutdownFuture.java renamed to driver-core/src/main/java/com/datastax/driver/core/CloseFuture.java driver-core/src/main/java/com/datastax/driver/core/ShutdownFuture.java renamed to driver-core/src/main/java/com/datastax/driver/core/CloseFuture.java

@@ -31,14 +31,14 @@
  * Note that this class implements <a href="http://code.google.com/p/guava-libraries/">Guava</a>'s {@code
  * ListenableFuture} and can so be used with Guava's future utilities.
  */
-public abstract class ShutdownFuture extends AbstractFuture<Void> {
+public abstract class CloseFuture extends AbstractFuture<Void> {
 
-    ShutdownFuture() {}
+    CloseFuture() {}
 
-    static ShutdownFuture immediateFuture() {
-        ShutdownFuture future = new ShutdownFuture() {
+    static CloseFuture immediateFuture() {
+        CloseFuture future = new CloseFuture() {
             @Override
-            public ShutdownFuture force() {
+            public CloseFuture force() {
                 return this;
             }
         };
@@ -47,26 +47,27 @@ public ShutdownFuture force() {
     }
 
     /**
-     * Try to force the completion of the shutdown this a future of.
+     * Try to force the completion of the shutdown this is a future of.
      * <p>
-     * This method will do its best to expedite the shutdown process. In particular, all connection
-     * will be closed right away, even if there is ongoing queries at the time of the call
-     * to this method.
+     * This method will do its best to expedite the shutdown process. In
+     * particular, all connections will be closed right away, even if there is
+     * ongoing queries at the time this method is called.
      * <p>
-     * Note that this method does not block. The completion of this method does not imply that the
-     * shutdown process and you still need to wait on this future to ensure that, though calling
-     * this method does ensure said future will return in a timely way.
+     * Note that this method does not block. The completion of this method does
+     * not imply the shutdown process is done, you still need to wait on this
+     * future to ensure that, but calling this method will ensure said
+     * future will return in a timely way.
      *
-     * @return this {@code ShutdownFuture}.
+     * @return this {@code CloseFuture}.
      */
-    public abstract ShutdownFuture force();
+    public abstract CloseFuture force();
 
     // Internal utility for cases where we want to build a future that wait on other ones
-    static class Forwarding extends ShutdownFuture {
+    static class Forwarding extends CloseFuture {
 
-        private final List<ShutdownFuture> futures;
+        private final List<CloseFuture> futures;
 
-        Forwarding(List<ShutdownFuture> futures) {
+        Forwarding(List<CloseFuture> futures) {
             this.futures = futures;
 
             Futures.addCallback(Futures.allAsList(futures), new FutureCallback<List<Void>>() {
@@ -81,8 +82,8 @@ public void onSuccess(List<Void> v) {
         }
 
         @Override
-        public ShutdownFuture force() {
-            for (ShutdownFuture future : futures)
+        public CloseFuture force() {
+            for (CloseFuture future : futures)
                 future.force();
             return this;
         }

driver-core/src/main/java/com/datastax/driver/core/Cluster.java

@@ -15,6 +15,7 @@
  */
 package com.datastax.driver.core;
 
+import java.io.Closeable;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.*;
@@ -55,7 +56,7 @@
  * the nodes currently in the cluster as well as new nodes joining the cluster
  * subsequently.
  */
-public class Cluster {
+public class Cluster implements Closeable {
 
     private static final Logger logger = LoggerFactory.getLogger(Cluster.class);
 
@@ -322,24 +323,40 @@ public Cluster unregister(LatencyTracker tracker) {
 
     /**
      * Initiates a shutdown of this cluster instance.
-     *
+     * <p>
      * This method is asynchronous and return a future on the completion
      * of the shutdown process. As soon a the cluster is shutdown, no
      * new request will be accepted, but already submitted queries are
-     * allowed to complete. Shutdown closes all connections from all
+     * allowed to complete. This method closes all connections from all
      * sessions and reclaims all ressources used by this Cluster
      * instance.
      * <p>
      * If for some reason you wish to expedite this process, the
-     * {@link ShutdownFuture#force} can be called on the result future.
+     * {@link CloseFuture#force} can be called on the result future.
      * <p>
-     * This method has no particular effect if the cluster was already shut
-     * down (in which case the returned future will return immediately).
+     * This method has no particular effect if the cluster was already closed
+     * (in which case the returned future will return immediately).
      *
-     * @return a future on the completion of the shtudown process.
+     * @return a future on the completion of the shutdown process.
+     */
+    public CloseFuture closeAsync() {
+        return manager.close();
+    }
+
+    /**
+     * Initiates a shutdown of this cluster instance and blocks until
+     * that shutdown completes.
+     * <p>
+     * This method is a shortcut for {@code closeAsync().get()}.
      */
-    public ShutdownFuture shutdown() {
-        return manager.shutdown();
+    public void close() {
+        try {
+            closeAsync().get();
+        } catch (ExecutionException e) {
+            throw DefaultResultSetFuture.extractCauseFromExecutionException(e);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
     }
 
     /**
@@ -842,7 +859,7 @@ class Manager implements Host.StateListener, Connection.DefaultResponseHandler {
         // An executor for tasks that migth block some time, like creating new connection, but are generally not too critical.
         final ListeningExecutorService blockingTasksExecutor;
 
-        final AtomicReference<ShutdownFuture> shutdownFuture = new AtomicReference<ShutdownFuture>();
+        final AtomicReference<CloseFuture> closeFuture = new AtomicReference<CloseFuture>();
 
         // All the queries that have been prepared (we keep them so we can re-prepared them when a node fail or a
         // new one join the cluster).
@@ -899,7 +916,7 @@ private void init() {
             try {
                 controlConnection.connect();
             } catch (NoHostAvailableException e) {
-                shutdown();
+                close();
                 throw e;
             }
         }
@@ -930,13 +947,13 @@ void reportLatency(Host host, long latencyNanos) {
             }
         }
 
-        boolean isShutdown() {
-            return shutdownFuture.get() != null;
+        boolean isClosed() {
+            return closeFuture.get() != null;
         }
 
-        private ShutdownFuture shutdown() {
+        private CloseFuture close() {
 
-            ShutdownFuture future = shutdownFuture.get();
+            CloseFuture future = closeFuture.get();
             if (future != null)
                 return future;
 
@@ -953,24 +970,24 @@ private ShutdownFuture shutdown() {
                 metrics.shutdown();
 
             // Then we shutdown all connections
-            List<ShutdownFuture> futures = new ArrayList<ShutdownFuture>(sessions.size() + 1);
-            futures.add(controlConnection.shutdown());
+            List<CloseFuture> futures = new ArrayList<CloseFuture>(sessions.size() + 1);
+            futures.add(controlConnection.closeAsync());
             for (Session session : sessions)
-                futures.add(session.shutdown());
+                futures.add(session.closeAsync());
 
-            future = new ClusterShutdownFuture(futures);
+            future = new ClusterCloseFuture(futures);
 
             // The rest will happen asynchonrously, when all connections are successfully closed
-            return shutdownFuture.compareAndSet(null, future)
+            return closeFuture.compareAndSet(null, future)
                  ? future
-                 : shutdownFuture.get(); // We raced, it's ok, return the future that was actually set
+                 : closeFuture.get(); // We raced, it's ok, return the future that was actually set
         }
 
         @Override
         public void onUp(final Host host) {
             logger.trace("Host {} is UP", host);
 
-            if (isShutdown())
+            if (isClosed())
                 return;
 
             if (host.isUp())
@@ -1043,7 +1060,7 @@ public void onDown(final Host host) {
         public void onDown(final Host host, final boolean isHostAddition) {
             logger.trace("Host {} is DOWN", host);
 
-            if (isShutdown())
+            if (isClosed())
                 return;
 
             // Note: we don't want to skip that method if !host.isUp() because we set isUp
@@ -1104,7 +1121,7 @@ protected boolean onUnknownException(Exception e, long nextDelayMs) {
         public void onAdd(final Host host) {
             logger.trace("Adding new host {}", host);
 
-            if (isShutdown())
+            if (isClosed())
                 return;
 
             try {
@@ -1154,7 +1171,7 @@ public void onFailure(Throwable t) {
 
         @Override
         public void onRemove(Host host) {
-            if (isShutdown())
+            if (isClosed())
                 return;
 
             host.setDown();
@@ -1267,7 +1284,7 @@ private void prepareAllQueries(Host host) throws InterruptedException {
                         }
                     }
                 } finally {
-                    connection.close();
+                    connection.closeAsync();
                 }
             } catch (ConnectionException e) {
                 // Ignore, not a big deal
@@ -1419,14 +1436,14 @@ private int delayForEvent(ProtocolEvent event) {
             return 0;
         }
 
-        private class ClusterShutdownFuture extends ShutdownFuture.Forwarding {
+        private class ClusterCloseFuture extends CloseFuture.Forwarding {
 
-            ClusterShutdownFuture(List<ShutdownFuture> futures) {
+            ClusterCloseFuture(List<CloseFuture> futures) {
                 super(futures);
             }
 
             @Override
-            public ShutdownFuture force() {
+            public CloseFuture force() {
                 reconnectionExecutor.shutdownNow();
                 scheduledTasksExecutor.shutdownNow();
                 executor.shutdownNow();

driver-core/src/main/java/com/datastax/driver/core/Connection.java

@@ -69,7 +69,7 @@ class Connection {
     private volatile boolean isDefunct;
     private volatile ConnectionException exception;
 
-    private final AtomicReference<ConnectionShutdownFuture> shutdownFuture = new AtomicReference<ConnectionShutdownFuture>();
+    private final AtomicReference<ConnectionCloseFuture> closeFuture = new AtomicReference<ConnectionCloseFuture>();
 
     /**
      * Create a new connection to a Cassandra node.
@@ -193,7 +193,7 @@ ConnectionException defunct(ConnectionException e) {
         exception = e;
         isDefunct = true;
         dispatcher.errorOutAllHandler(e);
-        close();
+        closeAsync();
         return e;
     }
 
@@ -311,16 +311,16 @@ public void operationComplete(ChannelFuture writeFuture) {
     }
 
     public boolean isClosed() {
-        return shutdownFuture.get() != null;
+        return closeFuture.get() != null;
     }
 
-    public ShutdownFuture close() {
+    public CloseFuture closeAsync() {
 
-        ConnectionShutdownFuture future = new ConnectionShutdownFuture();
-        if (!shutdownFuture.compareAndSet(null, future))
+        ConnectionCloseFuture future = new ConnectionCloseFuture();
+        if (!closeFuture.compareAndSet(null, future))
         {
-            // Shutdown had already been called, return the existing future
-            return shutdownFuture.get();
+            // close had already been called, return the existing future
+            return closeFuture.get();
         }
 
         logger.trace("[{}] closing connection", name);
@@ -510,10 +510,10 @@ public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
                 handler.cancelTimeout();
                 handler.callback.onSet(Connection.this, response, System.nanoTime() - handler.startTime);
 
-                // If we happen to be shutdown and we're the last outstanding request, we need to signal the shutdown future
+                // If we happen to be closed and we're the last outstanding request, we need to signal the close future
                 // (note: this is racy as the signaling can be called more than once, but that's not a problem)
                 if (isClosed() && pending.isEmpty())
-                    shutdownFuture.get().force();
+                    closeFuture.get().force();
             }
         }
 
@@ -563,10 +563,10 @@ public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) {
         }
     }
 
-    private class ConnectionShutdownFuture extends ShutdownFuture {
+    private class ConnectionCloseFuture extends CloseFuture {
 
         @Override
-        public ConnectionShutdownFuture force() {
+        public ConnectionCloseFuture force() {
             // Note: we must not call releaseExternalResources on the bootstrap, because this shutdown the executors, which are shared
 
             // This method can be thrown during Connection ctor, at which point channel is not yet set. This is ok.
@@ -584,9 +584,9 @@ public ConnectionShutdownFuture force() {
             future.addListener(new ChannelFutureListener() {
                 public void operationComplete(ChannelFuture future) {
                     if (future.getCause() != null)
-                        ConnectionShutdownFuture.this.setException(future.getCause());
+                        ConnectionCloseFuture.this.setException(future.getCause());
                     else
-                        ConnectionShutdownFuture.this.set(null);
+                        ConnectionCloseFuture.this.set(null);
                 }
             });
             return this;

driver-core/src/main/java/com/datastax/driver/core/ControlConnection.java

@@ -78,12 +78,12 @@ public void connect() {
         setNewConnection(reconnectInternal());
     }
 
-    public ShutdownFuture shutdown() {
+    public CloseFuture closeAsync() {
         // We don't have to be fancy here. We just set a flag so that we stop trying to reconnect (and thus change the
         // connection used) and shutdown the current one.
         isShutdown = true;
         Connection connection = connectionRef.get();
-        return connection == null ? ShutdownFuture.immediateFuture() : connection.close();
+        return connection == null ? CloseFuture.immediateFuture() : connection.closeAsync();
     }
 
     private void reconnect() {
@@ -146,7 +146,7 @@ private void setNewConnection(Connection newConnection) {
         logger.debug("[Control connection] Successfully connected to {}", newConnection.address);
         Connection old = connectionRef.getAndSet(newConnection);
         if (old != null && !old.isClosed())
-            old.close();
+            old.closeAsync();
     }
 
     private Connection reconnectInternal() {