DefaultRedisSlave implements RedisSlave { @Override public String info() { String info = ""; long lag = System.currentTimeMillis() - replAckTime; info = String.format( "ip=%s,port=%d,state=%s,offset=%d,lag=%d,remotePort=%d" , getClientIpAddress() == null ? ip() : getClientIpAddress(), getSlaveListeningPort(), slaveState != null ? slaveState.getDesc() : "null", replAckOff, lag/1000, remotePort()); return info; } DefaultRedisSlave(RedisClient redisClient); @Override void waitForRdbDumping(); @Override SLAVE_STATE getSlaveState(); @Override void ack(Long ackOff); @Override ChannelFuture writeFile(ReferenceFileRegion referenceFileRegion); @Override Long getAck(); @Override Long getAckTime(); @Override void beginWriteRdb(EofType eofType, long rdbFileOffset); @Override void rdbWriteComplete(); @Override void beginWriteCommands(long beginOffset); @Override ChannelFuture onCommand(ReferenceFileRegion referenceFileRegion); @Override String info(); @Override String ip(); @Override PARTIAL_STATE partialState(); @Override void partialSync(); @Override void processPsyncSequentially(Runnable runnable); @Override void markPsyncProcessed(); @Override String metaInfo(); @Override boolean isOpen(); @Override void close(); @Override void beforeCommand(); void addObserver(Observer observer); void removeObserver(Observer observer); RedisSlave becomeSlave(); RedisKeeperServer getRedisKeeperServer(); void setSlaveListeningPort(int port); int getSlaveListeningPort(); @Override void setClientIpAddress(String host); @Override String getClientIpAddress(); void capa(CAPA capa); @Override Set<CAPA> getCapas(); String[] readCommands(ByteBuf byteBuf); Channel channel(); void sendMessage(ByteBuf byteBuf); void sendMessage(byte[] bytes); void addChannelCloseReleaseResources(Releasable releasable); @Override void setClientEndpoint(Endpoint endpoint); @Override Endpoint getClientEndpoint(); @Override String toString(); @Override void release(); @Override boolean capaOf(CAPA capa); @Override boolean isKeeper(); @Override void setKeeper(); static final String KEY_RDB_DUMP_MAX_WAIT_MILLI; }
@Test public void testFuture() { SettableFuture<Boolean> objectSettableFuture = SettableFuture.create(); AtomicInteger listenerCount = new AtomicInteger(0); AtomicInteger notifyCount = new AtomicInteger(); executors.execute(new AbstractExceptionLogTask() { @Override protected void doRun() throws Exception { while (!Thread.interrupted()) { listenerCount.incrementAndGet(); objectSettableFuture.addListener(new Runnable() { @Override public void run() { notifyCount.incrementAndGet(); } }, MoreExecutors.directExecutor()); } logger.info("exit thread"); } }); sleep(10); objectSettableFuture.set(true); executors.shutdownNow(); sleep(10); logger.info("{}, {}", listenerCount, notifyCount); Assert.assertEquals(listenerCount.get(), notifyCount.get()); }