@Test public void testClientTimeout() throws Exception { Map<String, String> conf = ImmutableMap.<String,String>builder() .putAll(emptyConfig) .build(); RpcServer server = autoClose(new RpcServer(conf)); String secret = server.createSecret(); try { autoClose(server.registerClient("client", secret, new TestDispatcher(), 1L).get()); fail("Server should have timed out client."); } catch (ExecutionException ee) { assertTrue(ee.getCause() instanceof TimeoutException); } NioEventLoopGroup eloop = new NioEventLoopGroup(); Future<Rpc> clientRpcFuture = Rpc.createClient(conf, eloop, "localhost", server.getPort(), "client", secret, new TestDispatcher()); try { autoClose(clientRpcFuture.get()); fail("Client should have failed to connect to server."); } catch (ExecutionException ee) { assertFalse(ee.getCause() instanceof TimeoutException); } }
public static Promise<Rpc> createClient( Map<String, String> config, final NioEventLoopGroup eloop, String host, int port, final String clientId, final String secret, final RpcDispatcher dispatcher) throws Exception { final RpcConfiguration rpcConf = new RpcConfiguration(config); int connectTimeoutMs = (int) rpcConf.getConnectTimeoutMs(); final ChannelFuture cf = new Bootstrap() .group(eloop) .handler(new ChannelInboundHandlerAdapter() { }) .channel(NioSocketChannel.class) .option(ChannelOption.SO_KEEPALIVE, true) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs) .connect(host, port); final Promise<Rpc> promise = eloop.next().newPromise(); final AtomicReference<Rpc> rpc = new AtomicReference<Rpc>(); final Runnable timeoutTask = new Runnable() { @Override public void run() { promise.setFailure(new TimeoutException("Timed out waiting for RPC server connection.")); } }; final ScheduledFuture<?> timeoutFuture = eloop.schedule(timeoutTask, rpcConf.getServerConnectTimeoutMs(), TimeUnit.MILLISECONDS); cf.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture cf) throws Exception { if (cf.isSuccess()) { SaslClientHandler saslHandler = new SaslClientHandler(rpcConf, clientId, promise, timeoutFuture, secret, dispatcher); Rpc rpc = createRpc(rpcConf, saslHandler, (SocketChannel) cf.channel(), eloop); saslHandler.rpc = rpc; saslHandler.sendHello(cf.channel()); } else { promise.setFailure(cf.cause()); } } }); promise.addListener(new GenericFutureListener<Promise<Rpc>>() { @Override public void operationComplete(Promise<Rpc> p) { if (p.isCancelled()) { cf.cancel(true); } } }); return promise; }
Rpc implements Closeable { public static Promise<Rpc> createClient( Map<String, String> config, final NioEventLoopGroup eloop, String host, int port, final String clientId, final String secret, final RpcDispatcher dispatcher) throws Exception { final RpcConfiguration rpcConf = new RpcConfiguration(config); int connectTimeoutMs = (int) rpcConf.getConnectTimeoutMs(); final ChannelFuture cf = new Bootstrap() .group(eloop) .handler(new ChannelInboundHandlerAdapter() { }) .channel(NioSocketChannel.class) .option(ChannelOption.SO_KEEPALIVE, true) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs) .connect(host, port); final Promise<Rpc> promise = eloop.next().newPromise(); final AtomicReference<Rpc> rpc = new AtomicReference<Rpc>(); final Runnable timeoutTask = new Runnable() { @Override public void run() { promise.setFailure(new TimeoutException("Timed out waiting for RPC server connection.")); } }; final ScheduledFuture<?> timeoutFuture = eloop.schedule(timeoutTask, rpcConf.getServerConnectTimeoutMs(), TimeUnit.MILLISECONDS); cf.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture cf) throws Exception { if (cf.isSuccess()) { SaslClientHandler saslHandler = new SaslClientHandler(rpcConf, clientId, promise, timeoutFuture, secret, dispatcher); Rpc rpc = createRpc(rpcConf, saslHandler, (SocketChannel) cf.channel(), eloop); saslHandler.rpc = rpc; saslHandler.sendHello(cf.channel()); } else { promise.setFailure(cf.cause()); } } }); promise.addListener(new GenericFutureListener<Promise<Rpc>>() { @Override public void operationComplete(Promise<Rpc> p) { if (p.isCancelled()) { cf.cancel(true); } } }); return promise; } }
Rpc implements Closeable { public static Promise<Rpc> createClient( Map<String, String> config, final NioEventLoopGroup eloop, String host, int port, final String clientId, final String secret, final RpcDispatcher dispatcher) throws Exception { final RpcConfiguration rpcConf = new RpcConfiguration(config); int connectTimeoutMs = (int) rpcConf.getConnectTimeoutMs(); final ChannelFuture cf = new Bootstrap() .group(eloop) .handler(new ChannelInboundHandlerAdapter() { }) .channel(NioSocketChannel.class) .option(ChannelOption.SO_KEEPALIVE, true) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs) .connect(host, port); final Promise<Rpc> promise = eloop.next().newPromise(); final AtomicReference<Rpc> rpc = new AtomicReference<Rpc>(); final Runnable timeoutTask = new Runnable() { @Override public void run() { promise.setFailure(new TimeoutException("Timed out waiting for RPC server connection.")); } }; final ScheduledFuture<?> timeoutFuture = eloop.schedule(timeoutTask, rpcConf.getServerConnectTimeoutMs(), TimeUnit.MILLISECONDS); cf.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture cf) throws Exception { if (cf.isSuccess()) { SaslClientHandler saslHandler = new SaslClientHandler(rpcConf, clientId, promise, timeoutFuture, secret, dispatcher); Rpc rpc = createRpc(rpcConf, saslHandler, (SocketChannel) cf.channel(), eloop); saslHandler.rpc = rpc; saslHandler.sendHello(cf.channel()); } else { promise.setFailure(cf.cause()); } } }); promise.addListener(new GenericFutureListener<Promise<Rpc>>() { @Override public void operationComplete(Promise<Rpc> p) { if (p.isCancelled()) { cf.cancel(true); } } }); return promise; } private  Rpc(RpcConfiguration config, Channel channel, EventExecutorGroup egroup); }
Rpc implements Closeable { public static Promise<Rpc> createClient( Map<String, String> config, final NioEventLoopGroup eloop, String host, int port, final String clientId, final String secret, final RpcDispatcher dispatcher) throws Exception { final RpcConfiguration rpcConf = new RpcConfiguration(config); int connectTimeoutMs = (int) rpcConf.getConnectTimeoutMs(); final ChannelFuture cf = new Bootstrap() .group(eloop) .handler(new ChannelInboundHandlerAdapter() { }) .channel(NioSocketChannel.class) .option(ChannelOption.SO_KEEPALIVE, true) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs) .connect(host, port); final Promise<Rpc> promise = eloop.next().newPromise(); final AtomicReference<Rpc> rpc = new AtomicReference<Rpc>(); final Runnable timeoutTask = new Runnable() { @Override public void run() { promise.setFailure(new TimeoutException("Timed out waiting for RPC server connection.")); } }; final ScheduledFuture<?> timeoutFuture = eloop.schedule(timeoutTask, rpcConf.getServerConnectTimeoutMs(), TimeUnit.MILLISECONDS); cf.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture cf) throws Exception { if (cf.isSuccess()) { SaslClientHandler saslHandler = new SaslClientHandler(rpcConf, clientId, promise, timeoutFuture, secret, dispatcher); Rpc rpc = createRpc(rpcConf, saslHandler, (SocketChannel) cf.channel(), eloop); saslHandler.rpc = rpc; saslHandler.sendHello(cf.channel()); } else { promise.setFailure(cf.cause()); } } }); promise.addListener(new GenericFutureListener<Promise<Rpc>>() { @Override public void operationComplete(Promise<Rpc> p) { if (p.isCancelled()) { cf.cancel(true); } } }); return promise; } private  Rpc(RpcConfiguration config, Channel channel, EventExecutorGroup egroup); static Promise<Rpc> createClient(
      Map<String, String> config,
      final NioEventLoopGroup eloop,
      String host,
      int port,
      final String clientId,
      final String secret,
      final RpcDispatcher dispatcher); void addListener(Listener l); Future<Void> call(Object msg); Future<T> call(Object msg, Class<T> retType); Promise<T> createPromise(); @Override void close(); }
Rpc implements Closeable { public static Promise<Rpc> createClient( Map<String, String> config, final NioEventLoopGroup eloop, String host, int port, final String clientId, final String secret, final RpcDispatcher dispatcher) throws Exception { final RpcConfiguration rpcConf = new RpcConfiguration(config); int connectTimeoutMs = (int) rpcConf.getConnectTimeoutMs(); final ChannelFuture cf = new Bootstrap() .group(eloop) .handler(new ChannelInboundHandlerAdapter() { }) .channel(NioSocketChannel.class) .option(ChannelOption.SO_KEEPALIVE, true) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs) .connect(host, port); final Promise<Rpc> promise = eloop.next().newPromise(); final AtomicReference<Rpc> rpc = new AtomicReference<Rpc>(); final Runnable timeoutTask = new Runnable() { @Override public void run() { promise.setFailure(new TimeoutException("Timed out waiting for RPC server connection.")); } }; final ScheduledFuture<?> timeoutFuture = eloop.schedule(timeoutTask, rpcConf.getServerConnectTimeoutMs(), TimeUnit.MILLISECONDS); cf.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture cf) throws Exception { if (cf.isSuccess()) { SaslClientHandler saslHandler = new SaslClientHandler(rpcConf, clientId, promise, timeoutFuture, secret, dispatcher); Rpc rpc = createRpc(rpcConf, saslHandler, (SocketChannel) cf.channel(), eloop); saslHandler.rpc = rpc; saslHandler.sendHello(cf.channel()); } else { promise.setFailure(cf.cause()); } } }); promise.addListener(new GenericFutureListener<Promise<Rpc>>() { @Override public void operationComplete(Promise<Rpc> p) { if (p.isCancelled()) { cf.cancel(true); } } }); return promise; } private  Rpc(RpcConfiguration config, Channel channel, EventExecutorGroup egroup); static Promise<Rpc> createClient(
      Map<String, String> config,
      final NioEventLoopGroup eloop,
      String host,
      int port,
      final String clientId,
      final String secret,
      final RpcDispatcher dispatcher); void addListener(Listener l); Future<Void> call(Object msg); Future<T> call(Object msg, Class<T> retType); Promise<T> createPromise(); @Override void close();  }