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

@Override
public RFuture<Boolean> deleteAsync() {
    final RPromise<Boolean> result = new RedissonPromise<Boolean>();
    RFuture<Long> deleteFuture = redisson.getKeys().deleteAsync(
            requestQueueName, statusName, tasksCounterName, schedulerQueueName, tasksName);
    deleteFuture.addListener(new FutureListener<Long>() {
        @Override
        public void operationComplete(io.netty.util.concurrent.Future<Long> future) throws Exception {
            if (!future.isSuccess()) {
                result.tryFailure(future.cause());
                return;
            }

            result.trySuccess(future.getNow() > 0);
        }
    });
    return result;
}

private <T> void addListener(final RemotePromise<T> result) {
    result.getAddFuture().addListener(new FutureListener<Boolean>() {

        @Override
        public void operationComplete(io.netty.util.concurrent.Future<Boolean> future) throws Exception {
            if (!future.isSuccess()) {
                result.tryFailure(future.cause());
                return;
            }

            if (!future.getNow()) {
                result.tryFailure(new RejectedExecutionException("Task rejected. ExecutorService is in shutdown state"));
            }

        }
    });
}

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 addListener(RFuture<Long> startTimeFuture) {
    startTimeFuture.addListener(new FutureListener<Long>() {
        @Override
        public void operationComplete(io.netty.util.concurrent.Future<Long> future) throws Exception {
            if (!future.isSuccess()) {
                if (future.cause() instanceof RedissonShutdownException) {
                    return;
                }
                log.error(future.cause().getMessage(), future.cause());
                scheduleTask(System.currentTimeMillis() + 5 * 1000L);
                return;
            }

            if (future.getNow() != null) {
                scheduleTask(future.getNow());
            }
        }
    });
}

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);
            }
        }
    });
}

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 RFuture<Void> add(final ClientConnectionsEntry entry) {
    final RPromise<Void> result = connectionManager.newPromise();
    FutureListener<Void> listener = new FutureListener<Void>() {
        AtomicInteger counter = new AtomicInteger(2);
        @Override
        public void operationComplete(Future<Void> future) throws Exception {
            if (!future.isSuccess()) {
                result.tryFailure(future.cause());
                return;
            }
            if (counter.decrementAndGet() == 0) {
                String addr = entry.getClient().getIpAddr();
                ip2Entry.put(addr, entry);
                result.trySuccess(null);
            }
        }
    };

    RFuture<Void> slaveFuture = slaveConnectionPool.add(entry);
    slaveFuture.addListener(listener);
    RFuture<Void> pubSubFuture = pubSubConnectionPool.add(entry);
    pubSubFuture.addListener(listener);
    return result;
}

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 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);
        }
    });
}

@Override
public RFuture<Boolean> forceUnlockAsync() {
    RFuture<Boolean> result = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
          "if (redis.call('hget', KEYS[1], 'mode') == 'write') 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 <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;
}

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);
                }
            });
        }
    });
}

/**
 * Disconnect from the AudioConnect server and shutdown this client.<br>
 * If this client has already been or is being shutdown, this will do nothing.<br>
 * <b>Note:</b> This client instance will no longer be able to connect after this is called.
 * @return a Future for when this client has completely disconnected and been shutdown.
 */
public Future<?> shutdown() {
    if (bootstrap.group().isShuttingDown()) {
        return GlobalEventExecutor.INSTANCE.newSucceededFuture(null);
    }

    final Promise<Object> shutdownPromise = GlobalEventExecutor.INSTANCE.newPromise();

    disconnect().addListener(new FutureListener<Object>() {

        @Override
        public void operationComplete(Future<Object> future) {
            bootstrap.group().shutdownGracefully().addListener(new PromiseNotifier<>(shutdownPromise));
        }
    });

    return shutdownPromise;
}

private void handleSet(LightPayload payload, final Message.AddressedMessage original) {
    final Key<EdimaxPlugSwitch> key = new Key<>(EdimaxPlugSwitch.class, payload.getModule().getName());
    final EdimaxPlugSwitch plugSwitch = requireComponent(key);
    final boolean setOn = payload.getOn();

    plugSwitch.setOnAsync(setOn).addListener(new FutureListener<Boolean>() {
        @Override
        public void operationComplete(Future<Boolean> future) throws Exception {
            if (future.isSuccess()) {
                if (future.get() == Boolean.TRUE) {
                    replyStatus(original, setOn);
                } else {
                    sendReply(original, new Message(new ErrorPayload("Cannot switch light")));
                }
            } else {
                sendReply(original, new Message(new ErrorPayload(future.cause())));
            }
        }
    });
}

/**
 * Sends a "OpenDoor" message to the master.
 */
public void unlatchDoor() {
    final String door = getDoor();
    if (door == null) {
        Log.e(TAG, "Could not open the door. No door buzzer installed");
        fireUnlatchActionFinished(false);
        return;
    }

    DoorPayload doorPayload = new DoorPayload(door);
    final Message.AddressedMessage messageToMaster = sendMessageToMaster(MASTER_DOOR_UNLATCH, new Message(doorPayload));
    final Future<Void> messagePayloadFuture = newResponseFuture(messageToMaster);
    messagePayloadFuture.addListener(new FutureListener<Void>() {
        @Override
        public void operationComplete(Future<Void> future) throws Exception {
            fireUnlatchActionFinished(future.isSuccess());
        }
    });
}

private <R, T> void retryReadRandomAsync(final RedisCommand<T> command, final Promise<R> mainPromise,
        final List<Integer> slots, final Object... params) {
    final Promise<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 (slots.isEmpty()) {
                        mainPromise.setSuccess(null);
                    } else {
                        retryReadRandomAsync(command, mainPromise, slots, params);
                    }
                } else {
                    mainPromise.setSuccess(future.getNow());
                }
            } else {
                mainPromise.setFailure(future.cause());
            }
        }
    });

    Integer slot = slots.remove(0);
    async(true, slot, null, connectionManager.getCodec(), command, params, attemptPromise, 0);
}

@Override
public Future<Boolean> addAllAsync(final Collection<? extends V> c) {
    final Promise<Boolean> promise = newPromise();
    if (c.isEmpty()) {
        promise.setSuccess(false);
        return promise;
    }
    final int listSize = size();
    List<Object> args = new ArrayList<Object>(c.size() + 1);
    args.add(getName());
    args.addAll(c);
    Future<Long> res = commandExecutor.writeAsync(getName(), RPUSH, args.toArray());
    res.addListener(new FutureListener<Long>() {
        @Override
        public void operationComplete(Future<Long> future) throws Exception {
            if (future.isSuccess()) {
                promise.setSuccess(listSize != future.getNow());
            } else {
                promise.setFailure(future.cause());
            }
        }
    });
    return promise;
}

private void testAsync(byte[] key, byte[] value) {
    final Semaphore concurrencyControl = new Semaphore(pipeline);
    FutureListener<Object> listener = new FutureListener<Object>() {

        @Override
        public void operationComplete(Future<Object> future) throws Exception {
            if (future.isSuccess()) {
                concurrencyControl.release();
                reqCount.incrementAndGet();
            } else {
                future.cause().printStackTrace();
                System.exit(1);
            }
        }
    };
    while (!stop.get()) {
        concurrencyControl.acquireUninterruptibly(2);
        client.set(key, value).addListener(listener);
        client.get(key).addListener(listener);
    }
}

public static PromiseConverter<byte[]> toBytes(EventExecutor executor) {
    return new PromiseConverter<byte[]>(executor) {

        @Override
        public FutureListener<Object> newListener(final Promise<byte[]> promise) {
            return new FutureListener<Object>() {

                @Override
                public void operationComplete(Future<Object> future) throws Exception {
                    if (future.isSuccess()) {
                        Object resp = future.getNow();
                        if (resp instanceof RedisResponseException) {
                            promise.tryFailure((RedisResponseException) resp);
                        } else if (resp == RedisResponseDecoder.NULL_REPLY) {
                            promise.trySuccess(null);
                        } else {
                            promise.trySuccess((byte[]) resp);
                        }
                    } else {
                        promise.tryFailure(future.cause());
                    }
                }
            };
        }
    };
}

public static PromiseConverter<Double> toDouble(EventExecutor executor) {
    return new PromiseConverter<Double>(executor) {

        @Override
        public FutureListener<Object> newListener(final Promise<Double> promise) {
            return new FutureListener<Object>() {

                @Override
                public void operationComplete(Future<Object> future) throws Exception {
                    if (future.isSuccess()) {
                        Object resp = future.getNow();
                        if (resp instanceof RedisResponseException) {
                            promise.tryFailure((RedisResponseException) resp);
                        } else if (resp == RedisResponseDecoder.NULL_REPLY) {
                            promise.trySuccess(null);
                        } else {
                            promise.trySuccess(bytesToDouble((byte[]) resp));
                        }
                    } else {
                        promise.tryFailure(future.cause());
                    }
                }
            };
        }
    };
}

public static PromiseConverter<Long> toLong(EventExecutor executor) {
    return new PromiseConverter<Long>(executor) {

        @Override
        public FutureListener<Object> newListener(final Promise<Long> promise) {
            return new FutureListener<Object>() {

                @Override
                public void operationComplete(Future<Object> future) throws Exception {
                    if (future.isSuccess()) {
                        Object resp = future.getNow();
                        if (resp instanceof RedisResponseException) {
                            promise.tryFailure((RedisResponseException) resp);
                        } else if (resp == RedisResponseDecoder.NULL_REPLY) {
                            promise.trySuccess(null);
                        } else {
                            promise.trySuccess((Long) resp);
                        }
                    } else {
                        promise.tryFailure(future.cause());
                    }
                }
            };
        }
    };
}

public static PromiseConverter<Object> toObject(EventExecutor executor) {
    return new PromiseConverter<Object>(executor) {

        @Override
        public FutureListener<Object> newListener(final Promise<Object> promise) {
            return new FutureListener<Object>() {

                @Override
                public void operationComplete(Future<Object> future) throws Exception {
                    if (future.isSuccess()) {
                        Object resp = future.getNow();
                        if (resp instanceof RedisResponseException) {
                            promise.tryFailure((RedisResponseException) resp);
                        } else if (resp == RedisResponseDecoder.NULL_REPLY) {
                            promise.trySuccess(null);
                        } else {
                            promise.trySuccess(resp);
                        }
                    } else {
                        promise.tryFailure(future.cause());
                    }
                }
            };
        }
    };
}

public static PromiseConverter<String> toString(EventExecutor executor) {
    return new PromiseConverter<String>(executor) {

        @Override
        public FutureListener<Object> newListener(final Promise<String> promise) {
            return new FutureListener<Object>() {

                @Override
                public void operationComplete(Future<Object> future) throws Exception {
                    if (future.isSuccess()) {
                        Object resp = future.getNow();
                        if (resp instanceof RedisResponseException) {
                            promise.tryFailure((RedisResponseException) resp);
                        } else if (resp == RedisResponseDecoder.NULL_REPLY) {
                            promise.trySuccess(null);
                        } else {
                            promise.trySuccess(resp.toString());
                        }
                    } else {
                        promise.tryFailure(future.cause());
                    }
                }
            };
        }
    };
}

public static PromiseConverter<Void> toVoid(EventExecutor executor) {
    return new PromiseConverter<Void>(executor) {

        @Override
        public FutureListener<Object> newListener(final Promise<Void> promise) {
            return new FutureListener<Object>() {

                @Override
                public void operationComplete(Future<Object> future) throws Exception {
                    if (future.isSuccess()) {
                        Object resp = future.getNow();
                        if (resp instanceof RedisResponseException) {
                            promise.tryFailure((RedisResponseException) resp);
                        } else {
                            promise.trySuccess(null);
                        }
                    } else {
                        promise.tryFailure(future.cause());
                    }
                }
            };
        }
    };
}

@SuppressWarnings({
    "unchecked", "rawtypes"
})
private void callBlocking(final NedisClient client, Method method, Object[] args,
        final long previousTimeoutMs, final Promise promise) throws IllegalAccessException,
        InvocationTargetException {
    ((Future) method.invoke(client, args)).addListener(new FutureListener() {

        @Override
        public void operationComplete(Future future) throws Exception {
            if (future.isSuccess()) {
                promise.trySuccess(future.getNow());
            } else {
                promise.tryFailure(future.cause());
            }
            resetTimeout(client, previousTimeoutMs);
        }
    });
}

private void setInfiniteTimeout(final NedisClient client, final Method method,
        final Object[] args, @SuppressWarnings("rawtypes") final Promise promise) {
    client.setTimeout(0L).addListener(new FutureListener<Long>() {

        @Override
        public void operationComplete(Future<Long> future) throws Exception {
            // will not fail, but could be null
            Long previousTimeoutMs = future.get();
            if (previousTimeoutMs == null) {
                promise.tryFailure(new IllegalStateException("already closed"));
            } else {
                callBlocking(client, method, args, previousTimeoutMs.longValue(), promise);
            }
        }
    });
}

@SuppressWarnings({
    "unchecked", "rawtypes"
})
private void call(final NedisClient client, Method method, Object[] args,
        final Promise promise) throws IllegalAccessException, InvocationTargetException {
    ((Future) method.invoke(client, args)).addListener(new FutureListener() {

        @Override
        public void operationComplete(Future future) throws Exception {
            if (future.isSuccess()) {
                promise.trySuccess(future.getNow());
            } else {
                promise.tryFailure(future.cause());
            }
            client.release();
        }
    });
}

private Future<NedisClient> newClient() {
    Future<NedisClientImpl> f = NedisClientBuilder.create().group(group).channel(channelClass)
            .timeoutMs(timeoutMs).belongTo(this).connect(remoteAddress);

    final Promise<NedisClient> promise = getEventExecutor(f).newPromise();
    f.addListener(new FutureListener<NedisClientImpl>() {

        @Override
        public void operationComplete(Future<NedisClientImpl> future) throws Exception {
            if (future.isSuccess()) {
                initialize(promise, future.getNow(), State.AUTH);
            } else {
                promise.tryFailure(future.cause());
            }
        }

    });
    return promise;
}

public Future<WriteResult> send(ChicagoMessage message, Promise<WriteResult> result) {
  // TODO schedule a timeout to fail this write
  resultMap.put(message.id, result);
  Future<Channel> channelResult = connectionPool.acquire();
  System.out.println("Acquiring Node");
  channelResult.addListener(new FutureListener<Channel>() {
    public void operationComplete(Future<Channel> future) {
      if (future.isSuccess()) {
        System.out.println("Node acquired!");
        Channel channel = future.getNow();
        // TODO could maybe put a listener here to track successful writes
        channel.writeAndFlush(message).addListener(new ChannelFutureListener() {
          public void operationComplete(ChannelFuture channelFuture) {
            System.out.println("write finished for " + message.id);
          }
        });
      } else {
        result.setFailure(future.cause());
      }
    }
  });
  return result;
}

private void acquireWithRetry(AsyncRetryLoop retry, DefaultPromise<Channel> result) {
  Future<Channel> poolResult = simpleChannelPool.acquire();
  poolResult.addListener(
      new FutureListener<Channel>() {
        public void operationComplete(Future<Channel> f) {
          if (f.isSuccess()) {
            result.setSuccess(f.getNow());
          } else {
            // deal with connection failure here.
            if (retry.canRetry()) {
              retry.attempt(() -> acquireWithRetry(retry, result));
            } else {
              result.setFailure(f.cause());
            }
          }
        }
      });
}

@Command(aliases = "load", desc = "Loads a new map and registers it to the system")
@Require("admin")
public void startgame(Platform platform, final CommandSender sender, Mappack mappack){
    final Future<Map> future = platform.getMapLoader().createMapFor(mappack);
    future.addListener(new FutureListener<Map>(){
        @Override
        public void operationComplete(Future<Map> mapFuture) throws Exception {
            //TODO: handle potential future failures
            ComponentBuilder builder = new ComponentBuilder("Map ").color(ChatColor.GREEN);
            //.event(new HoverEvent(HoverEvent.Action.SHOW_TEXT, new TextComponent("Click to go to this map")))
            //.event(new ClickEvent(ClickEvent.Action.RUN_COMMAND, "/goto " + future.get().defaultWorld().getDimensionId)) //TODO: teleport to default world
            builder.append(future.get().mappack().getMetadata().name()).append(" was loaded");
            sender.sendMessage(builder.create());
        }
    });
}

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 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 <T, R> RFuture<R> async(long timeout, Codec encoder, RedisCommand<T> command, Object ... params) {
    final RPromise<R> promise = new RedissonPromise<R>();
    if (timeout == -1) {
        timeout = redisClient.getCommandTimeout();
    }

    if (redisClient.getEventLoopGroup().isShuttingDown()) {
        RedissonShutdownException cause = new RedissonShutdownException("Redisson is shutdown");
        return RedissonPromise.newFailedFuture(cause);
    }

    final ScheduledFuture<?> scheduledFuture = redisClient.getEventLoopGroup().schedule(new Runnable() {
        @Override
        public void run() {
            RedisTimeoutException ex = new RedisTimeoutException("Command execution timeout for " + redisClient.getAddr());
            promise.tryFailure(ex);
        }
    }, timeout, TimeUnit.MILLISECONDS);

    promise.addListener(new FutureListener<R>() {
        @Override
        public void operationComplete(Future<R> future) throws Exception {
            scheduledFuture.cancel(false);
        }
    });
    send(new CommandData<T, R>(promise, encoder, command, params));
    return promise;
}