AbstractRedisMasterReplication extends AbstractLifecycle implements RedisMasterReplication { protected void connectWithMaster() { if (!(getLifecycleState().isStarting() || getLifecycleState().isStarted())) { logger.info("[connectWithMaster][do not connect, is stopped!!]{}", redisMaster.masterEndPoint()); return; } Bootstrap b = new Bootstrap(); b.group(nioEventLoopGroup).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true) .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) .handler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) throws Exception { ChannelPipeline p = ch.pipeline(); p.addLast(new LoggingHandler(LogLevel.DEBUG)); p.addLast(new NettySimpleMessageHandler()); p.addLast(createHandler()); } }); doConnect0(b); } 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 testListeningPortCommandTimeOut() throws Exception { System.setProperty(KEY_MASTER_CONNECT_RETRY_DELAY_SECONDS, "0"); Server server = startEmptyServer(); ProxyConnectProtocol protocol = new DefaultProxyConnectProtocolParser().read("PROXY ROUTE TCP: ProxyEnabledEndpoint endpoint = new ProxyEnabledEndpoint("127.0.0.1", server.getPort(), protocol); when(redisMaster.masterEndPoint()).thenReturn(endpoint); KeeperResourceManager proxyEndpointManager = getRegistry().getComponent(KeeperResourceManager.class); ProxyEndpointSelector selector = mock(ProxyEndpointSelector.class); when(proxyEndpointManager.createProxyEndpointSelector(any())).thenReturn(selector); when(selector.nextHop()).thenReturn(protocol.nextEndpoints().get(0)); DefaultRedisMasterReplication.PROXYED_REPLICATION_COMMAND_TIMEOUT_MILLI = 5; redisMasterReplication = new DefaultRedisMasterReplication(redisMaster, redisKeeperServer, nioEventLoopGroup, scheduled, replTimeoutMilli, proxyEndpointManager); redisMasterReplication = spy(redisMasterReplication); redisMasterReplication.initialize(); redisMasterReplication.start(); verify(redisMasterReplication, timeout(500).atLeast(2)).connectWithMaster(); }