AbstractRedisMasterReplication extends AbstractLifecycle implements RedisMasterReplication { private void checkKeeper() { Replconf replconf = new Replconf(clientPool, ReplConfType.KEEPER, scheduled, commandTimeoutMilli); executeCommand(replconf).addListener(new CommandFutureListener<Object>() { @Override public void operationComplete(CommandFuture<Object> commandFuture) throws Exception { if(commandFuture.isSuccess()){ redisMaster.setKeeper(); } } }); } AbstractRedisMasterReplication(RedisKeeperServer redisKeeperServer, RedisMaster redisMaster, NioEventLoopGroup nioEventLoopGroup,
										  ScheduledExecutorService scheduled, int replTimeoutMilli, KeeperResourceManager resourceManager);  AbstractRedisMasterReplication(RedisKeeperServer redisKeeperServer, RedisMaster redisMaster, NioEventLoopGroup nioEventLoopGroup,
										  ScheduledExecutorService scheduled, KeeperResourceManager resourceManager); RedisMaster getRedisMaster(); void startReplication(); @Override void handleResponse(Channel channel, ByteBuf byteBuf); @SuppressWarnings({ "unchecked", "rawtypes" }) @Override void masterConnected(Channel channel); @Override void masterDisconntected(Channel channel); @Override boolean canSendPsync(); @Override void updateReplicationObserver(RedisMasterReplicationObserver observer); @Override String toString(); @Override void onFullSync(); @Override void reFullSync(); @Override void beginWriteRdb(EofType eofType, long masterRdbOffset); @Override void endWriteRdb(); @Override void onContinue(String requestReplId, String responseReplId); void setRdbDumper(RdbDumper dumper); RdbDumper getRdbDumper(); @Override RedisMaster redisMaster(); static String KEY_MASTER_CONNECT_RETRY_DELAY_SECONDS; static String KEY_REPLICATION_TIMEOUT; static int DEFAULT_REPLICATION_TIMEOUT_MILLI; static final int REPLCONF_INTERVAL_MILLI; static final int PSYNC_RETRY_INTERVAL_MILLI; }
@Test public void testCheckKeeper() throws Exception { Server server = startServer("-ERR syntax error"); SimpleObjectPool<NettyClient> pool = getXpipeNettyClientKeyedObjectPool().getKeyPool(new DefaultProxyEndpoint("127.0.0.1", server.getPort())); Replconf replconf = new Replconf(pool, Replconf.ReplConfType.KEEPER, scheduled, 100); replconf.execute().addListener(new CommandFutureListener<Object>() { @Override public void operationComplete(CommandFuture<Object> commandFuture) throws Exception { if (!commandFuture.isSuccess()) { logger.error("[]", commandFuture.cause()); Assert.assertTrue(commandFuture.cause() instanceof RedisError); } } }); replconf.future().await(); server.stop(); }