Agree with https://stackoverflow.com/users/3371051/pedro, CompletableFuture is Java Built-in, while ListenableFuture is counting on Guava. It's easy convert ListenableFuture to CompletableFuture.
- Create CompletableFuture, complete() or completeExceptionally() in ListenableFuture listenable call back.
- use thenCompose to chain CompletableFuture with ListenableFuture
- use allOf to replace Futures.allAsList
Like the following tested change I did for com.datastax driver core 3.11.0
CompletableFuture initAsync() {
CompletableFuture ret = new CompletableFuture();
if (factory.isShutdown) {
ret.completeExceptionally(
new ConnectionException(endPoint, "Connection factory is shut down"));
return ret;
}
ProtocolVersion protocolVersion =
factory.protocolVersion == null
? ProtocolVersion.NEWEST_SUPPORTED
: factory.protocolVersion;
try {
Bootstrap bootstrap = factory.newBootstrap();
ProtocolOptions protocolOptions = factory.configuration.getProtocolOptions();
bootstrap.handler(
new Initializer(
this,
protocolVersion,
protocolOptions.getCompression().compressor(),
protocolOptions.getSSLOptions(),
factory.configuration.getPoolingOptions().getHeartbeatIntervalSeconds(),
factory.configuration.getNettyOptions(),
factory.configuration.getCodecRegistry(),
factory.configuration.getMetricsOptions().isEnabled()
? factory.manager.metrics
: null));
ChannelFuture future = bootstrap.connect(endPoint.resolve());
writer.incrementAndGet();
future.addListener(
new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
writer.decrementAndGet();
// Note: future.channel() can be null in some error cases, so we need to guard against
// it in the rest of the code below.
channel = future.channel();
if (isClosed() && channel != null) {
channel
.close()
.addListener(
new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
ret.completeExceptionally(
new TransportException(
Connection.this.endPoint,
"Connection closed during initialization."));
}
});
} else {
if (channel != null) {
Connection.this.factory.allChannels.add(channel);
}
if (!future.isSuccess()) {
if (logger.isDebugEnabled())
logger.debug(
String.format(
"%s Error connecting to %s%s",
Connection.this,
Connection.this.endPoint,
extractMessage(future.cause())));
ret.completeExceptionally(
new TransportException(
Connection.this.endPoint, "Cannot connect", future.cause()));
} else {
assert channel != null;
logger.debug(
"{} Connection established, initializing transport", Connection.this);
channel.closeFuture().addListener(new ChannelCloseListener());
ret.complete(null);
}
}
}
});
} catch (RuntimeException e) {
closeAsync().force();
throw e;
}
Executor initExecutor =
factory.manager.configuration.getPoolingOptions().getInitializationExecutor();
return ret.thenCompose(
Void -> {
CompletableFuture ret2 = new CompletableFuture();
ProtocolOptions protocolOptions = factory.configuration.getProtocolOptions();
Future startupResponseFuture =
write(
new Requests.Startup(
protocolOptions.getCompression(), protocolOptions.isNoCompact()));
ListenableFuture channelReadyFuture =
GuavaCompatibility.INSTANCE.transformAsync(
startupResponseFuture,
onStartupResponse(protocolVersion, initExecutor),
initExecutor);
GuavaCompatibility.INSTANCE.addCallback(
channelReadyFuture,
new FutureCallback() {
@Override
public void onSuccess(Void result) {
ret2.complete(null);
}
@Override
public void onFailure(Throwable t) {
// Make sure the connection gets properly closed.
if (t instanceof ClusterNameMismatchException
|| t instanceof UnsupportedProtocolVersionException) {
// Just propagate
closeAsync().force();
ret2.completeExceptionally(t);
} else {
// Defunct to ensure that the error will be signaled (marking the host down)
Throwable e =
(t instanceof ConnectionException
|| t instanceof DriverException
|| t instanceof InterruptedException
|| t instanceof Error)
? t
: new ConnectionException(
Connection.this.endPoint,
String.format(
"Unexpected error during transport initialization (%s)", t),
t);
ret2.completeExceptionally(defunct(e));
}
// Ensure the connection gets closed if the caller cancels the returned future.
if (!isClosed()) {
closeAsync().force();
}
}
});
return ret2;
});
}
private CompletableFuture createPools(Collection hosts) {
List> futures = Lists.newArrayListWithCapacity(hosts.size());
for (Host host : hosts)
if (host.state != Host.State.DOWN) futures.add(maybeAddPool(host, null));
return CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
}
