future : futures) future.setException(t);
}
- });
+ },
+ MoreExecutors.directExecutor());
}
// If we didn't dequeue all events (or new ones arrived since we did), make sure we eventually
diff --git a/driver-core/src/main/java/com/datastax/driver/core/GuavaCompatibility.java b/driver-core/src/main/java/com/datastax/driver/core/GuavaCompatibility.java
deleted file mode 100644
index 069f550b049..00000000000
--- a/driver-core/src/main/java/com/datastax/driver/core/GuavaCompatibility.java
+++ /dev/null
@@ -1,361 +0,0 @@
-/*
- * Copyright DataStax, Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.datastax.driver.core;
-
-import com.datastax.driver.core.exceptions.DriverInternalError;
-import com.google.common.base.Function;
-import com.google.common.collect.BiMap;
-import com.google.common.collect.Maps;
-import com.google.common.net.HostAndPort;
-import com.google.common.reflect.TypeToken;
-import com.google.common.util.concurrent.AsyncFunction;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
-import java.lang.reflect.ParameterizedType;
-import java.lang.reflect.Type;
-import java.util.Map;
-import java.util.concurrent.Executor;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A compatibility layer to support a wide range of Guava versions.
- *
- * The driver is compatible with Guava 16.0.1 or higher, but Guava 20 introduced incompatible
- * breaking changes in its API, that could in turn be breaking for legacy driver clients if we
- * simply upgraded our dependency. We don't want to increment our major version "just" for Guava (we
- * have other changes planned).
- *
- *
Therefore we depend on Guava 19, which has both the deprecated and the new APIs, and detect
- * the actual version at runtime in order to call the relevant methods.
- *
- *
This is a hack, and might not work with subsequent Guava releases; the real fix is to stop
- * exposing Guava in our public API. We'll address that in version 4 of the driver.
- */
-@SuppressWarnings("deprecation")
-public abstract class GuavaCompatibility {
-
- private static final Logger logger = LoggerFactory.getLogger(GuavaCompatibility.class);
-
- /**
- * The unique instance of this class, that is compatible with the Guava version found in the
- * classpath.
- */
- public static final GuavaCompatibility INSTANCE = selectImplementation();
-
- /**
- * Force the initialization of the class. This should be called early to ensure a fast failure if
- * an incompatible version of Guava is in the classpath (the driver code calls it when loading the
- * {@link Cluster} class).
- */
- public static void init() {
- // nothing to do, we just want the static initializers to run
- }
-
- /**
- * Returns a {@code Future} whose result is taken from the given primary {@code input} or, if the
- * primary input fails, from the {@code Future} provided by the {@code fallback}.
- *
- * @see Futures#withFallback(ListenableFuture, com.google.common.util.concurrent.FutureFallback)
- * @see Futures#catchingAsync(ListenableFuture, Class, AsyncFunction)
- */
- public abstract ListenableFuture withFallback(
- ListenableFuture input, AsyncFunction fallback);
-
- /**
- * Returns a {@code Future} whose result is taken from the given primary {@code input} or, if the
- * primary input fails, from the {@code Future} provided by the {@code fallback}.
- *
- * @see Futures#withFallback(ListenableFuture, com.google.common.util.concurrent.FutureFallback,
- * Executor)
- * @see Futures#catchingAsync(ListenableFuture, Class, AsyncFunction, Executor)
- */
- public abstract ListenableFuture withFallback(
- ListenableFuture input, AsyncFunction fallback, Executor executor);
-
- /**
- * Registers separate success and failure callbacks to be run when the {@code Future}'s
- * computation is {@linkplain java.util.concurrent.Future#isDone() complete} or, if the
- * computation is already complete, immediately.
- *
- * The callback is run in {@link #sameThreadExecutor()}.
- *
- * @see Futures#addCallback(ListenableFuture, FutureCallback, Executor)
- */
- public void addCallback(ListenableFuture input, FutureCallback callback) {
- addCallback(input, callback, sameThreadExecutor());
- }
-
- /**
- * Registers separate success and failure callbacks to be run when the {@code Future}'s
- * computation is {@linkplain java.util.concurrent.Future#isDone() complete} or, if the
- * computation is already complete, immediately.
- *
- * @see Futures#addCallback(ListenableFuture, FutureCallback, Executor)
- */
- public void addCallback(
- ListenableFuture input, FutureCallback callback, Executor executor) {
- Futures.addCallback(input, callback, executor);
- }
-
- /**
- * Returns a new {@code ListenableFuture} whose result is the product of applying the given {@code
- * Function} to the result of the given {@code Future}.
- *
- *
The callback is run in {@link #sameThreadExecutor()}.
- *
- * @see Futures#transform(ListenableFuture, Function, Executor)
- */
- public ListenableFuture transform(
- ListenableFuture input, Function function) {
- return transform(input, function, sameThreadExecutor());
- }
-
- /**
- * Returns a new {@code ListenableFuture} whose result is the product of applying the given {@code
- * Function} to the result of the given {@code Future}.
- *
- * @see Futures#transform(ListenableFuture, Function, Executor)
- */
- public ListenableFuture transform(
- ListenableFuture input, Function function, Executor executor) {
- return Futures.transform(input, function, executor);
- }
-
- /**
- * Returns a new {@code ListenableFuture} whose result is asynchronously derived from the result
- * of the given {@code Future}. More precisely, the returned {@code Future} takes its result from
- * a {@code Future} produced by applying the given {@code AsyncFunction} to the result of the
- * original {@code Future}.
- *
- * @see Futures#transform(ListenableFuture, AsyncFunction)
- * @see Futures#transformAsync(ListenableFuture, AsyncFunction)
- */
- public abstract ListenableFuture transformAsync(
- ListenableFuture input, AsyncFunction function);
-
- /**
- * Returns a new {@code ListenableFuture} whose result is asynchronously derived from the result
- * of the given {@code Future}. More precisely, the returned {@code Future} takes its result from
- * a {@code Future} produced by applying the given {@code AsyncFunction} to the result of the
- * original {@code Future}.
- *
- * @see Futures#transform(ListenableFuture, AsyncFunction, Executor)
- * @see Futures#transformAsync(ListenableFuture, AsyncFunction, Executor)
- */
- public abstract ListenableFuture transformAsync(
- ListenableFuture input, AsyncFunction function, Executor executor);
-
- /**
- * Returns true if {@code target} is a supertype of {@code argument}. "Supertype" is defined
- * according to the rules for type arguments introduced with Java generics.
- *
- * @see TypeToken#isAssignableFrom(Type)
- * @see TypeToken#isSupertypeOf(Type)
- */
- public abstract boolean isSupertypeOf(TypeToken target, TypeToken argument);
-
- /**
- * Returns an {@link Executor} that runs each task in the thread that invokes {@link
- * Executor#execute execute}, as in {@link
- * java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy}.
- *
- * @see MoreExecutors#sameThreadExecutor()
- * @see MoreExecutors#directExecutor()
- */
- public abstract Executor sameThreadExecutor();
-
- /**
- * Returns the portion of the given {@link HostAndPort} instance that should represent the
- * hostname or IPv4/IPv6 literal.
- *
- * The method {@code HostAndPort.getHostText} has been replaced with {@code
- * HostAndPort.getHost} starting with Guava 20.0; it has been completely removed in Guava 22.0.
- */
- @SuppressWarnings("JavaReflectionMemberAccess")
- public String getHost(HostAndPort hostAndPort) {
- try {
- // Guava >= 20.0
- return (String) HostAndPort.class.getMethod("getHost").invoke(hostAndPort);
- } catch (Exception e) {
- // Guava < 22.0
- return hostAndPort.getHostText();
- }
- }
-
- private static GuavaCompatibility selectImplementation() {
- if (isGuava_19_0_OrHigher()) {
- logger.info("Detected Guava >= 19 in the classpath, using modern compatibility layer");
- return new Version19OrHigher();
- } else if (isGuava_16_0_1_OrHigher()) {
- logger.info("Detected Guava < 19 in the classpath, using legacy compatibility layer");
- return new Version18OrLower();
- } else {
- throw new DriverInternalError(
- "Detected incompatible version of Guava in the classpath. "
- + "You need 16.0.1 or higher.");
- }
- }
-
- private static class Version18OrLower extends GuavaCompatibility {
-
- @Override
- public ListenableFuture withFallback(
- ListenableFuture input, final AsyncFunction fallback) {
- return Futures.withFallback(
- input,
- new com.google.common.util.concurrent.FutureFallback() {
- @Override
- public ListenableFuture create(Throwable t) throws Exception {
- return fallback.apply(t);
- }
- });
- }
-
- @Override
- public ListenableFuture withFallback(
- ListenableFuture input,
- final AsyncFunction fallback,
- Executor executor) {
- return Futures.withFallback(
- input,
- new com.google.common.util.concurrent.FutureFallback() {
- @Override
- public ListenableFuture create(Throwable t) throws Exception {
- return fallback.apply(t);
- }
- },
- executor);
- }
-
- @Override
- public ListenableFuture transformAsync(
- ListenableFuture input, AsyncFunction function) {
- return Futures.transform(input, function);
- }
-
- @Override
- public ListenableFuture transformAsync(
- ListenableFuture input,
- AsyncFunction function,
- Executor executor) {
- return Futures.transform(input, function, executor);
- }
-
- @Override
- public boolean isSupertypeOf(TypeToken target, TypeToken argument) {
- return target.isAssignableFrom(argument);
- }
-
- @Override
- public Executor sameThreadExecutor() {
- return MoreExecutors.sameThreadExecutor();
- }
- }
-
- private static class Version19OrHigher extends GuavaCompatibility {
-
- @Override
- public ListenableFuture withFallback(
- ListenableFuture input, AsyncFunction fallback) {
- return withFallback(input, fallback, sameThreadExecutor());
- }
-
- @Override
- public ListenableFuture withFallback(
- ListenableFuture input,
- AsyncFunction fallback,
- Executor executor) {
- return Futures.catchingAsync(input, Throwable.class, fallback, executor);
- }
-
- @Override
- public ListenableFuture transformAsync(
- ListenableFuture input, AsyncFunction function) {
- return transformAsync(input, function, sameThreadExecutor());
- }
-
- @Override
- public ListenableFuture transformAsync(
- ListenableFuture input,
- AsyncFunction function,
- Executor executor) {
- return Futures.transformAsync(input, function, executor);
- }
-
- @Override
- public boolean isSupertypeOf(TypeToken target, TypeToken argument) {
- return target.isSupertypeOf(argument);
- }
-
- @Override
- public Executor sameThreadExecutor() {
- return MoreExecutors.directExecutor();
- }
- }
-
- private static boolean isGuava_19_0_OrHigher() {
- return methodExists(
- Futures.class,
- "transformAsync",
- ListenableFuture.class,
- AsyncFunction.class,
- Executor.class);
- }
-
- private static boolean isGuava_16_0_1_OrHigher() {
- // Cheap check for < 16.0
- if (!methodExists(Maps.class, "asConverter", BiMap.class)) {
- return false;
- }
- // More elaborate check to filter out 16.0, which has a bug in TypeToken. We need 16.0.1.
- boolean resolved = false;
- TypeToken> mapOfString = TypeTokens.mapOf(String.class, String.class);
- Type type = mapOfString.getType();
- if (type instanceof ParameterizedType) {
- ParameterizedType pType = (ParameterizedType) type;
- Type[] types = pType.getActualTypeArguments();
- if (types.length == 2) {
- TypeToken valueType = TypeToken.of(types[1]);
- resolved = valueType.getRawType().equals(String.class);
- }
- }
- if (!resolved) {
- logger.debug(
- "Detected Guava issue #1635 which indicates that version 16.0 is in the classpath");
- }
- return resolved;
- }
-
- private static boolean methodExists(
- Class declaringClass, String methodName, Class... parameterTypes) {
- try {
- declaringClass.getMethod(methodName, parameterTypes);
- return true;
- } catch (Exception e) {
- logger.debug(
- "Error while checking existence of method "
- + declaringClass.getSimpleName()
- + "."
- + methodName,
- e);
- return false;
- }
- }
-}
diff --git a/driver-core/src/main/java/com/datastax/driver/core/HostConnectionPool.java b/driver-core/src/main/java/com/datastax/driver/core/HostConnectionPool.java
index 14954b3b5dc..a27a70de265 100644
--- a/driver-core/src/main/java/com/datastax/driver/core/HostConnectionPool.java
+++ b/driver-core/src/main/java/com/datastax/driver/core/HostConnectionPool.java
@@ -38,6 +38,7 @@
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import com.google.common.util.concurrent.Uninterruptibles;
import io.netty.util.concurrent.EventExecutor;
@@ -372,7 +373,7 @@ private void addCallback(
manager.cluster.manager.configuration.getPoolingOptions().getInitializationExecutor();
final ListenableFuture> allConnectionsFuture = Futures.allAsList(connectionFutures);
- GuavaCompatibility.INSTANCE.addCallback(
+ Futures.addCallback(
allConnectionsFuture,
new FutureCallback>() {
@Override
@@ -449,8 +450,9 @@ public void onFailure(Throwable t) {
private ListenableFuture handleErrors(
ListenableFuture connectionInitFuture, Executor executor) {
- return GuavaCompatibility.INSTANCE.withFallback(
+ return Futures.catchingAsync(
connectionInitFuture,
+ Throwable.class,
new AsyncFunction() {
@Override
public ListenableFuture apply(Throwable t) throws Exception {
@@ -708,7 +710,7 @@ private void dequeue(final Connection connection) {
} else {
// Otherwise the keyspace did need to be set, tie the pendingBorrow future to the set
// keyspace completion.
- GuavaCompatibility.INSTANCE.addCallback(
+ Futures.addCallback(
setKeyspaceFuture,
new FutureCallback() {
@@ -726,7 +728,8 @@ public void onFailure(Throwable t) {
pendingBorrow.setException(t);
connection.inFlight.decrementAndGet();
}
- });
+ },
+ MoreExecutors.directExecutor());
}
}
}
@@ -1033,7 +1036,7 @@ public void run() {
}
}
},
- GuavaCompatibility.INSTANCE.sameThreadExecutor());
+ MoreExecutors.directExecutor());
futures.add(future);
}
}
diff --git a/driver-core/src/main/java/com/datastax/driver/core/PoolingOptions.java b/driver-core/src/main/java/com/datastax/driver/core/PoolingOptions.java
index 3741ef97ec0..03b7c5e87b1 100644
--- a/driver-core/src/main/java/com/datastax/driver/core/PoolingOptions.java
+++ b/driver-core/src/main/java/com/datastax/driver/core/PoolingOptions.java
@@ -20,6 +20,7 @@
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.MoreExecutors;
import java.util.Map;
import java.util.concurrent.Executor;
@@ -134,8 +135,7 @@ public class PoolingOptions {
/** The default value for {@link #getHeartbeatIntervalSeconds()} ({@value}). */
public static final int DEFAULT_HEARTBEAT_INTERVAL_SECONDS = 30;
- private static final Executor DEFAULT_INITIALIZATION_EXECUTOR =
- GuavaCompatibility.INSTANCE.sameThreadExecutor();
+ private static final Executor DEFAULT_INITIALIZATION_EXECUTOR = MoreExecutors.directExecutor();
private volatile Cluster.Manager manager;
private volatile ProtocolVersion protocolVersion;
diff --git a/driver-core/src/main/java/com/datastax/driver/core/QueryOptions.java b/driver-core/src/main/java/com/datastax/driver/core/QueryOptions.java
index 3fa3e335b51..da3a4b0aaee 100644
--- a/driver-core/src/main/java/com/datastax/driver/core/QueryOptions.java
+++ b/driver-core/src/main/java/com/datastax/driver/core/QueryOptions.java
@@ -18,6 +18,8 @@
import com.datastax.driver.core.exceptions.UnsupportedFeatureException;
import com.datastax.driver.core.utils.MoreFutures;
import com.datastax.driver.core.utils.MoreObjects;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.MoreExecutors;
/** Options related to defaults for individual queries. */
public class QueryOptions {
@@ -335,7 +337,7 @@ public QueryOptions setMetadataEnabled(boolean enabled) {
// 1. call submitNodeListRefresh() first to
// be able to compute the token map for the first time,
// which will be incomplete due to the lack of keyspace metadata
- GuavaCompatibility.INSTANCE.addCallback(
+ Futures.addCallback(
manager.submitNodeListRefresh(),
new MoreFutures.SuccessCallback() {
@Override
@@ -345,7 +347,8 @@ public void onSuccess(Void result) {
// this time with information about keyspaces
manager.submitSchemaRefresh(null, null, null, null);
}
- });
+ },
+ MoreExecutors.directExecutor());
}
return this;
}
diff --git a/driver-core/src/main/java/com/datastax/driver/core/RequestHandler.java b/driver-core/src/main/java/com/datastax/driver/core/RequestHandler.java
index ba23b4bbb35..27f6b67d184 100644
--- a/driver-core/src/main/java/com/datastax/driver/core/RequestHandler.java
+++ b/driver-core/src/main/java/com/datastax/driver/core/RequestHandler.java
@@ -44,7 +44,9 @@
import com.google.common.collect.Iterators;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import java.nio.ByteBuffer;
@@ -470,7 +472,7 @@ private boolean query(final Host host) {
routingKey,
statementKeyspace,
statementTable);
- GuavaCompatibility.INSTANCE.addCallback(
+ Futures.addCallback(
connectionFuture,
new FutureCallback() {
@Override
@@ -522,7 +524,8 @@ public void onFailure(Throwable t) {
}
findNextHostAndQuery();
}
- });
+ },
+ MoreExecutors.directExecutor());
return true;
}
diff --git a/driver-core/src/main/java/com/datastax/driver/core/SessionManager.java b/driver-core/src/main/java/com/datastax/driver/core/SessionManager.java
index 54cceda6126..12cd2970394 100644
--- a/driver-core/src/main/java/com/datastax/driver/core/SessionManager.java
+++ b/driver-core/src/main/java/com/datastax/driver/core/SessionManager.java
@@ -38,6 +38,7 @@
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import com.google.common.util.concurrent.Uninterruptibles;
import java.nio.ByteBuffer;
@@ -99,7 +100,7 @@ public ListenableFuture initAsync() {
Collection hosts = cluster.getMetadata().allHosts();
ListenableFuture allPoolsCreatedFuture = createPools(hosts);
ListenableFuture allPoolsUpdatedFuture =
- GuavaCompatibility.INSTANCE.transformAsync(
+ Futures.transformAsync(
allPoolsCreatedFuture,
new AsyncFunction() {
@Override
@@ -108,9 +109,10 @@ public ListenableFuture