Java 类io.netty.util.concurrent.Future 实例源码

@Override
@SuppressForbidden(reason = "debug")
protected void stopInternal() {
    Releasables.close(serverOpenChannels, () -> {
        final List<Tuple<String, Future<?>>> serverBootstrapCloseFutures = new ArrayList<>(serverBootstraps.size());
        for (final Map.Entry<String, ServerBootstrap> entry : serverBootstraps.entrySet()) {
            serverBootstrapCloseFutures.add(
                Tuple.tuple(entry.getKey(), entry.getValue().config().group().shutdownGracefully(0, 5, TimeUnit.SECONDS)));
        }
        for (final Tuple<String, Future<?>> future : serverBootstrapCloseFutures) {
            future.v2().awaitUninterruptibly();
            if (!future.v2().isSuccess()) {
                logger.debug(
                    (Supplier<?>) () -> new ParameterizedMessage(
                        "Error closing server bootstrap for profile [{}]", future.v1()), future.v2().cause());
            }
        }
        serverBootstraps.clear();

        if (bootstrap != null) {
            bootstrap.config().group().shutdownGracefully(0, 5, TimeUnit.SECONDS).awaitUninterruptibly();
            bootstrap = null;
        }
    });
}

public RFuture<String> tryAcquireAsync() {
    final RPromise<String> result = newPromise();
    RFuture<String> res = tryAcquireAsync(1, nonExpirableTimeout);
    res.addListener(new FutureListener<String>() {
        @Override
        public void operationComplete(Future<String> future) throws Exception {
            if (!future.isSuccess()) {
                result.tryFailure(future.cause());
                return;
            }

            String permitId = future.getNow();
            if (permitId != null && !permitId.startsWith(":")) {
                if (!result.trySuccess(permitId)) {
                    releaseAsync(permitId);
                }
            } else {
                result.trySuccess(null);
            }
        }
    });
    return result;
}

@Override
public RFuture<Void> releaseAsync(final String permitId) {
    final RPromise<Void> result = newPromise();
    tryReleaseAsync(permitId).addListener(new FutureListener<Boolean>() {
        @Override
        public void operationComplete(Future<Boolean> future) throws Exception {
            if (!future.isSuccess()) {
                result.tryFailure(future.cause());
                return;
            }

            if (future.getNow()) {
                result.trySuccess(null);
            } else {
                result.tryFailure(new IllegalArgumentException("Permit with id " + permitId + " has already been released or doesn't exist"));
            }
        }
    });
    return result;
}

private void write(final byte[] value, final RPromise<Void> result, final int chunkSize, final int i) {
    final int len = Math.min(value.length - i*chunkSize, chunkSize);
    byte[] bytes = Arrays.copyOfRange(value, i*chunkSize, i*chunkSize + len);
    writeAsync(bytes).addListener(new FutureListener<Void>() {
        @Override
        public void operationComplete(Future<Void> future) throws Exception {
            if (!future.isSuccess()) {
                result.tryFailure(future.cause());
                return;
            }

            int j = i + 1;
            if (j*chunkSize > value.length) {
                result.trySuccess(null);
            } else {
                write(value, result, chunkSize, j);
            }
        }
    });
}

public void sendPacket(Packet packetIn, GenericFutureListener <? extends Future <? super Void >> listener, GenericFutureListener <? extends Future <? super Void >> ... listeners)
{
    if (this.isChannelOpen())
    {
        this.flushOutboundQueue();
        this.dispatchPacket(packetIn, (GenericFutureListener[])ArrayUtils.add(listeners, 0, listener));
    }
    else
    {
        this.field_181680_j.writeLock().lock();

        try
        {
            this.outboundPacketsQueue.add(new NetworkManager.InboundHandlerTuplePacketListener(packetIn, (GenericFutureListener[])ArrayUtils.add(listeners, 0, listener)));
        }
        finally
        {
            this.field_181680_j.writeLock().unlock();
        }
    }
}

/**
 * Kick a player from the server with a reason
 */
public void kickPlayerFromServer(String reason)
{
    final ChatComponentText chatcomponenttext = new ChatComponentText(reason);
    this.netManager.sendPacket(new S40PacketDisconnect(chatcomponenttext), new GenericFutureListener < Future <? super Void >> ()
    {
        public void operationComplete(Future <? super Void > p_operationComplete_1_) throws Exception
        {
            NetHandlerPlayServer.this.netManager.closeChannel(chatcomponenttext);
        }
    }, new GenericFutureListener[0]);
    this.netManager.disableAutoRead();
    Futures.getUnchecked(this.serverController.addScheduledTask(new Runnable()
    {
        public void run()
        {
            NetHandlerPlayServer.this.netManager.checkDisconnected();
        }
    }));
}

private RFuture<T> acquireConnection(RedisCommand<?> command, final ClientConnectionsEntry entry) {
    final RPromise<T> result = connectionManager.newPromise();

    AcquireCallback<T> callback = new AcquireCallback<T>() {
        @Override
        public void run() {
            result.removeListener(this);
            connectTo(entry, result);
        }

        @Override
        public void operationComplete(Future<T> future) throws Exception {
            entry.removeConnection(this);
        }
    };

    result.addListener(callback);
    acquireConnection(entry, callback);

    return result;
}

private void createConnection(final ClientConnectionsEntry entry, final RPromise<T> promise) {
    RFuture<T> connFuture = connect(entry);
    connFuture.addListener(new FutureListener<T>() {
        @Override
        public void operationComplete(Future<T> future) throws Exception {
            if (!future.isSuccess()) {
                promiseFailure(entry, promise, future.cause());
                return;
            }

            T conn = future.getNow();
            if (!conn.isActive()) {
                promiseFailure(entry, promise, conn);
                return;
            }

            connectedSuccessful(entry, promise, conn);
        }
    });
}

public void sendPacket(Packet packetIn, GenericFutureListener <? extends Future <? super Void >> listener, GenericFutureListener <? extends Future <? super Void >> ... listeners)
{
    if (this.isChannelOpen())
    {
        this.flushOutboundQueue();
        this.dispatchPacket(packetIn, (GenericFutureListener[])ArrayUtils.add(listeners, 0, listener));
    }
    else
    {
        this.field_181680_j.writeLock().lock();

        try
        {
            this.outboundPacketsQueue.add(new NetworkManager.InboundHandlerTuplePacketListener(packetIn, (GenericFutureListener[])ArrayUtils.add(listeners, 0, listener)));
        }
        finally
        {
            this.field_181680_j.writeLock().unlock();
        }
    }
}

private void psubscribe(final String channelName, final Collection<RedisPubSubListener<?>> listeners,
        final Codec subscribeCodec) {
    RFuture<PubSubConnectionEntry> subscribeFuture = connectionManager.psubscribe(channelName, subscribeCodec, null);
    subscribeFuture.addListener(new FutureListener<PubSubConnectionEntry>() {
        @Override
        public void operationComplete(Future<PubSubConnectionEntry> future)
                throws Exception {
            if (!future.isSuccess()) {
                psubscribe(channelName, listeners, subscribeCodec);
                return;
            }

            PubSubConnectionEntry newEntry = future.getNow();
            for (RedisPubSubListener<?> redisPubSubListener : listeners) {
                newEntry.addListener(channelName, redisPubSubListener);
            }
            log.debug("resubscribed listeners for '{}' channel-pattern", channelName);
        }
    });
}

@Override
public void run() {
    context.channelPool().acquire().addListener((Future<Channel> channelFuture) -> {
        if (channelFuture.isSuccess()) {
            try {
                channel = channelFuture.getNow();
                initializePerRequestHandlers();
                channel.attr(REQUEST_CONTEXT_KEY).set(context);
                channel.attr(RESPONSE_COMPLETE_KEY).set(false);
                makeRequest(context.nettyRequest());
            } catch (Exception e) {
                handleFailure(() -> "Failed to make request to " + endpoint(), e);
            }
        } else {
            handleFailure(() -> "Failed to create connection to " + endpoint(), channelFuture.cause());
        }
    });
}

@Override
public RFuture<Boolean> forceUnlockAsync() {
    RFuture<Boolean> result = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            "if (redis.call('hget', KEYS[1], 'mode') == 'read') then " +
                "redis.call('del', KEYS[1]); " +
                "redis.call('publish', KEYS[2], ARGV[1]); " +
                "return 1; " +
            "end; " +
            "return 0; ",
            Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage);

      result.addListener(new FutureListener<Boolean>() {
          @Override
          public void operationComplete(Future<Boolean> future) throws Exception {
              if (future.isSuccess() && future.getNow()) {
                  cancelExpirationRenewal();
              }
          }
      });

      return result;
}

private void checkClusterState(final ClusterServersConfig cfg, final Iterator<URI> iterator, final AtomicReference<Throwable> lastException) {
    if (!iterator.hasNext()) {
        log.error("Can't update cluster state", lastException.get());
        scheduleClusterChangeCheck(cfg, null);
        return;
    }
    if (!getShutdownLatch().acquire()) {
        return;
    }
    final URI uri = iterator.next();
    RFuture<RedisConnection> connectionFuture = connect(cfg, uri);
    connectionFuture.addListener(new FutureListener<RedisConnection>() {
        @Override
        public void operationComplete(Future<RedisConnection> future) throws Exception {
            if (!future.isSuccess()) {
                lastException.set(future.cause());
                getShutdownLatch().release();
                checkClusterState(cfg, iterator, lastException);
                return;
            }

            RedisConnection connection = future.getNow();
            updateClusterState(cfg, connection, iterator, uri);
        }
    });
}

private <R, T> void retryReadRandomAsync(final RedisCommand<T> command, final RPromise<R> mainPromise,
        final List<MasterSlaveEntry> nodes, final Object... params) {
    final RPromise<R> attemptPromise = connectionManager.newPromise();
    attemptPromise.addListener(new FutureListener<R>() {
        @Override
        public void operationComplete(Future<R> future) throws Exception {
            if (future.isSuccess()) {
                if (future.getNow() == null) {
                    if (nodes.isEmpty()) {
                        mainPromise.trySuccess(null);
                    } else {
                        retryReadRandomAsync(command, mainPromise, nodes, params);
                    }
                } else {
                    mainPromise.trySuccess(future.getNow());
                }
            } else {
                mainPromise.tryFailure(future.cause());
            }
        }
    });

    MasterSlaveEntry entry = nodes.remove(0);
    async(true, new NodeSource(entry), connectionManager.getCodec(), command, params, attemptPromise, 0);
}

protected <V, R> void releaseConnection(final NodeSource source, final RFuture<RedisConnection> connectionFuture,
        final boolean isReadOnly, RPromise<R> attemptPromise, final AsyncDetails<V, R> details) {
    attemptPromise.addListener(new FutureListener<R>() {
        @Override
        public void operationComplete(Future<R> future) throws Exception {
            if (!connectionFuture.isSuccess()) {
                return;
            }

            RedisConnection connection = connectionFuture.getNow();
            connectionManager.getShutdownLatch().release();
            if (isReadOnly) {
                connectionManager.releaseRead(source, connection);
            } else {
                connectionManager.releaseWrite(source, connection);
            }

            if (log.isDebugEnabled()) {
                log.debug("connection released for command {} and params {} from slot {} using connection {}",
                        details.getCommand(), Arrays.toString(details.getParams()), details.getSource(), connection);
            }
        }
    });
}

private RFuture<Boolean> tryAcquireOnceAsync(long leaseTime, TimeUnit unit, final long threadId) {
    if (leaseTime != -1) {
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
    }
    RFuture<Boolean> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
    ttlRemainingFuture.addListener(new FutureListener<Boolean>() {
        @Override
        public void operationComplete(Future<Boolean> future) throws Exception {
            if (!future.isSuccess()) {
                return;
            }

            Boolean ttlRemaining = future.getNow();
            // lock acquired
            if (ttlRemaining) {
                scheduleExpirationRenewal(threadId);
            }
        }
    });
    return ttlRemainingFuture;
}

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
    if (leaseTime != -1) {
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    ttlRemainingFuture.addListener(new FutureListener<Long>() {
        @Override
        public void operationComplete(Future<Long> future) throws Exception {
            if (!future.isSuccess()) {
                return;
            }

            Long ttlRemaining = future.getNow();
            // lock acquired
            if (ttlRemaining == null) {
                scheduleExpirationRenewal(threadId);
            }
        }
    });
    return ttlRemainingFuture;
}

public void sendPacket(Packet packetIn)
{
    if (this.isChannelOpen())
    {
        this.flushOutboundQueue();
        this.dispatchPacket(packetIn, (GenericFutureListener <? extends Future <? super Void >> [])null);
    }
    else
    {
        this.field_181680_j.writeLock().lock();

        try
        {
            this.outboundPacketsQueue.add(new NetworkManager.InboundHandlerTuplePacketListener(packetIn, (GenericFutureListener[])null));
        }
        finally
        {
            this.field_181680_j.writeLock().unlock();
        }
    }
}

private void loadValue(final K key, final RPromise<V> result, final RLock lock,
        final long threadId) {
    commandExecutor.getConnectionManager().getExecutor().execute(new Runnable() {
        @Override
        public void run() {
            final V value = options.getLoader().load(key);
            if (value == null) {
                result.trySuccess(value);
                return;
            }

            externalPutAsync(key, value).addListener(new FutureListener<V>() {
                @Override
                public void operationComplete(Future<V> future) throws Exception {
                    if (!future.isSuccess()) {
                        lock.unlockAsync(threadId);
                        result.tryFailure(future.cause());
                        return;
                    }

                    unlock(result, lock, threadId, value);
                }
            });
        }
    });
}

@Override
public Future<Channel> getConnection(HostAndPort address)
{
    try {
        Bootstrap bootstrap = new Bootstrap()
                .group(group)
                .channel(NioSocketChannel.class)
                .option(CONNECT_TIMEOUT_MILLIS, saturatedCast(connectTimeout.toMillis()))
                .handler(new ThriftClientInitializer(
                        messageFraming,
                        messageEncoding,
                        requestTimeout,
                        socksProxy,
                        sslContextSupplier));

        Promise<Channel> promise = group.next().newPromise();
        bootstrap.connect(new InetSocketAddress(address.getHost(), address.getPort()))
                .addListener((ChannelFutureListener) future -> notifyConnect(future, promise));
        return promise;
    }
    catch (Throwable e) {
        return group.next().newFailedFuture(new TTransportException(e));
    }
}

public void sendPacket(Packet<?> packetIn)
{
    if (this.isChannelOpen())
    {
        this.flushOutboundQueue();
        this.dispatchPacket(packetIn, (GenericFutureListener <? extends Future <? super Void >> [])null);
    }
    else
    {
        this.readWriteLock.writeLock().lock();

        try
        {
            this.outboundPacketsQueue.add(new NetworkManager.InboundHandlerTuplePacketListener(packetIn, (GenericFutureListener[])null));
        }
        finally
        {
            this.readWriteLock.writeLock().unlock();
        }
    }
}

/**
 * Kick a player from the server with a reason
 */
public void kickPlayerFromServer(String reason)
{
    final ChatComponentText chatcomponenttext = new ChatComponentText(reason);
    this.netManager.sendPacket(new S40PacketDisconnect(chatcomponenttext), new GenericFutureListener < Future <? super Void >> ()
    {
        public void operationComplete(Future <? super Void > p_operationComplete_1_) throws Exception
        {
            NetHandlerPlayServer.this.netManager.closeChannel(chatcomponenttext);
        }
    }, new GenericFutureListener[0]);
    this.netManager.disableAutoRead();
    Futures.getUnchecked(this.serverController.addScheduledTask(new Runnable()
    {
        public void run()
        {
            NetHandlerPlayServer.this.netManager.checkDisconnected();
        }
    }));
}

public void sendPacket(Packet packetIn)
{
    if (this.isChannelOpen())
    {
        this.flushOutboundQueue();
        this.dispatchPacket(packetIn, (GenericFutureListener <? extends Future <? super Void >> [])null);
    }
    else
    {
        this.field_181680_j.writeLock().lock();

        try
        {
            this.outboundPacketsQueue.add(new NetworkManager.InboundHandlerTuplePacketListener(packetIn, (GenericFutureListener[])null));
        }
        finally
        {
            this.field_181680_j.writeLock().unlock();
        }
    }
}

public void sendPacket(Packet packetIn, GenericFutureListener <? extends Future <? super Void >> listener, GenericFutureListener <? extends Future <? super Void >> ... listeners)
{
    if (this.isChannelOpen())
    {
        this.flushOutboundQueue();
        this.dispatchPacket(packetIn, (GenericFutureListener[])ArrayUtils.add(listeners, 0, listener));
    }
    else
    {
        this.field_181680_j.writeLock().lock();

        try
        {
            this.outboundPacketsQueue.add(new NetworkManager.InboundHandlerTuplePacketListener(packetIn, (GenericFutureListener[])ArrayUtils.add(listeners, 0, listener)));
        }
        finally
        {
            this.field_181680_j.writeLock().unlock();
        }
    }
}

public void sendPacket(Packet packetIn)
{
    if (this.isChannelOpen())
    {
        this.flushOutboundQueue();
        this.dispatchPacket(packetIn, (GenericFutureListener <? extends Future <? super Void >> [])null);
    }
    else
    {
        this.field_181680_j.writeLock().lock();

        try
        {
            this.outboundPacketsQueue.add(new NetworkManager.InboundHandlerTuplePacketListener(packetIn, (GenericFutureListener[])null));
        }
        finally
        {
            this.field_181680_j.writeLock().unlock();
        }
    }
}

public void sendPacket(Packet packetIn, GenericFutureListener <? extends Future <? super Void >> listener, GenericFutureListener <? extends Future <? super Void >> ... listeners)
{
    if (this.isChannelOpen())
    {
        this.flushOutboundQueue();
        this.dispatchPacket(packetIn, (GenericFutureListener[])ArrayUtils.add(listeners, 0, listener));
    }
    else
    {
        this.field_181680_j.writeLock().lock();

        try
        {
            this.outboundPacketsQueue.add(new NetworkManager.InboundHandlerTuplePacketListener(packetIn, (GenericFutureListener[])ArrayUtils.add(listeners, 0, listener)));
        }
        finally
        {
            this.field_181680_j.writeLock().unlock();
        }
    }
}

@SuppressWarnings("AmbiguousMixinReference")
@Redirect(method = "sendPacket", at = @At(value = "INVOKE", target = "net/minecraft/network/NetworkManager.dispatchPacket(Lnet/minecraft/network/Packet;[Lio/netty/util/concurrent/GenericFutureListener;)V"))
private void sendPacket(NetworkManager networkManager, Packet<?> packetIn, @Nullable final GenericFutureListener<? extends Future<?super Void>>[] futureListeners) {
    PacketEvent event = new PacketEvent.Send(packetIn);
    ClientAPI.EVENT_BUS.post(event);
    if (event.isCancelled())
        return;

    this.dispatchPacket(event.getPacket(), null);
}

@Override
public void operationComplete(Future<Channel> cf) throws Exception {

    if (cf.isCancelled()) {
        promise.cancel(true);
        return;
    }

    if (!cf.isSuccess()) {
        promise.completeExceptionally(cf.cause());
        return;
    }

    Channel channel = cf.getNow();
    promise.whenComplete((result, error) -> pool.release(channel));

    try {

        FastdfsOperation<T> fastdfsOperation = new FastdfsOperation<>(channel, requester, replier, promise);
        if (LOG.isDebugEnabled()) {
            LOG.debug("execute {}", fastdfsOperation);
        }

        fastdfsOperation.execute();
    } catch (Exception e) {
        promise.completeExceptionally(e);
    }
}

private void startListenerHandle(Future future, SuccessfulListener listener) {
    try {
        Thread.sleep(10);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    if (!future.isSuccess()) future.cause().printStackTrace();
    if (listener != null) listener.onSuccess(future.isSuccess());
}

private void awaitResultAsync(final RemoteInvocationOptions optionsCopy, final RemotePromise<Object> result,
        final RemoteServiceRequest request, final String responseName, final String ackName) {
    RFuture<Boolean> deleteFuture = redisson.getBucket(ackName).deleteAsync();
    deleteFuture.addListener(new FutureListener<Boolean>() {
        @Override
        public void operationComplete(Future<Boolean> future) throws Exception {
            if (!future.isSuccess()) {
                result.tryFailure(future.cause());
                return;
            }

            awaitResultAsync(optionsCopy, result, request, responseName);
        }
    });
}

public MapWriterPromise(RFuture<R> f, final CommandAsyncExecutor commandExecutor, final MapWriterTask<R> task) {
    f.addListener(new FutureListener<R>() {
        @Override
        public void operationComplete(final Future<R> future) throws Exception {
            if (!future.isSuccess()) {
                tryFailure(future.cause());
                return;
            }

            if (task.condition(future)) {
                commandExecutor.getConnectionManager().getExecutor().execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            task.execute();
                        } catch (Exception e) {
                            tryFailure(e);
                            return;
                        }
                        trySuccess(future.getNow());
                    }
                });
            } else {
                trySuccess(future.getNow());
            }
        }
    });
}

@Override
public void operationComplete(Future<R> future) throws Exception {
    if (future.isSuccess() && task.condition(future)) {
        enqueueRunnable(new Runnable() {
            @Override
            public void run() {
                try {
                    task.execute();
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        });
    }
}

@Override
public void operationComplete(Future<T> future) throws Exception {
    if (!future.isSuccess()) {
        promise.tryFailure(future.cause());
        return;
    }

    promise.trySuccess(future.getNow());
}

@Override
 public void subscribe(final Subscriber<? super T> subscriber) {
     try {
         subscriber.onSubscribe(new ReactiveSubscription<T>(this, subscriber) {

             @Override
             protected void onRequest(long n) {
                 supplier.get().addListener(new FutureListener<T>() {
                     @Override
                     public void operationComplete(Future<T> future) throws Exception {
                         if (!future.isSuccess()) {
                             onError(future.cause());
                             return;
                         }

                         if (future.getNow() != null) {
                             onNext(future.getNow());
                         }
                         onComplete();
                     }
                 });
             }
         });
     } catch (Throwable throwable) {
//Exceptions.throwIfFatal(throwable);
         Exceptions.getFinalCause(throwable);
         subscriber.onError(throwable);
     }
 }

public RFuture<Void> add(final ClientConnectionsEntry entry) {
    final RPromise<Void> promise = connectionManager.newPromise();
    promise.addListener(new FutureListener<Void>() {
        @Override
        public void operationComplete(Future<Void> future) throws Exception {
            entries.add(entry);
        }
    });
    initConnections(entry, promise, true);
    return promise;
}

private void subscribe(final String channelName, final Collection<RedisPubSubListener<?>> listeners,
        final Codec subscribeCodec) {
    RFuture<PubSubConnectionEntry> subscribeFuture = connectionManager.subscribe(subscribeCodec, channelName, listeners.toArray(new RedisPubSubListener[listeners.size()]));
    subscribeFuture.addListener(new FutureListener<PubSubConnectionEntry>() {

        @Override
        public void operationComplete(Future<PubSubConnectionEntry> future)
                throws Exception {
            if (future.isSuccess()) {
                log.debug("resubscribed listeners of '{}' channel to {}", channelName, future.getNow().getConnection().getRedisClient());
            }
        }
    });
}

@Override
public void operationComplete(Future<Object> future) throws Exception {
    if (!future.isSuccess()) {
        connection.closeAsync();
        connectionPromise.tryFailure(future.cause());
        return;
    }
    if (commandsCounter.decrementAndGet() == 0) {
        connectionPromise.trySuccess(connection);
    }
}

private void subscribe(final String channelName, final RPromise<PubSubConnectionEntry> promise,
        final PubSubType type, final AsyncSemaphore lock, final PubSubConnectionEntry connEntry,
        final RedisPubSubListener<?>... listeners) {
    for (RedisPubSubListener<?> listener : listeners) {
        connEntry.addListener(channelName, listener);
    }
    connEntry.getSubscribeFuture(channelName, type).addListener(new FutureListener<Void>() {
        @Override
        public void operationComplete(Future<Void> future) throws Exception {
            lock.release();
            promise.trySuccess(connEntry);
        }
    });
}

@Override
public void operationComplete(Future<? super Void> future) throws Exception {
  if (!future.isSuccess()) {
    logger.error("Unable to maintain connection {}.  Closing connection.", connection.getName());
    connection.close();
  }
}

@Override
public void operationComplete(Future<Channel> cf) throws Exception {

    if (cf.isCancelled()) {
        promise.cancel(true);
        return;
    }

    if (!cf.isSuccess()) {
        promise.completeExceptionally(cf.cause());
        return;
    }

    Channel channel = cf.getNow();
    promise.whenComplete((result, error) -> pool.release(channel));

    try {

        FastdfsOperation<T> fastdfsOperation = new FastdfsOperation<>(channel, requestor,
            replier, promise);
        if (LOG.isDebugEnabled()) {
            LOG.debug("execute {}", fastdfsOperation);
        }

        fastdfsOperation.execute();
    } catch (Exception e) {
        promise.completeExceptionally(e);
    }
}