@Test(timeout = 40000) public void testSendCloseSession() throws Exception { Assert.assertTrue( "Didn't instantiate ServerCnxnFactory with NettyServerCnxnFactory!", serverFactory instanceof NettyServerCnxnFactory); final ZooKeeper zk = createClient(); final ZooKeeperServer zkServer = getServer(serverFactory); final String path = "/a"; try { zk.create(path, "test".getBytes(StandardCharsets.UTF_8), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); Assert.assertNotNull("Didn't create znode:" + path, zk.exists(path, true)); Assert.assertEquals(1, zkServer.getZKDatabase().getDataTree().getWatchCount()); Iterable<ServerCnxn> connections = serverFactory.getConnections(); Assert.assertEquals("Mismatch in number of live connections!", 1, serverFactory.getNumAliveConnections()); for (ServerCnxn serverCnxn : connections) { serverCnxn.sendCloseSession(); } LOG.info("Waiting for the channel disconnected event"); int timeout = 0; while (serverFactory.getNumAliveConnections() != 0) { Thread.sleep(1000); timeout += 1000; if (timeout > CONNECTION_TIMEOUT) { Assert.fail("The number of live connections should be 0"); } } Assert.assertEquals(0, zkServer.getZKDatabase().getDataTree().getWatchCount()); } finally { zk.close(); } }
@Override public void sendCloseSession() { sendBuffer(ServerCnxnFactory.closeConn); }
NettyServerCnxn extends ServerCnxn { @Override public void sendCloseSession() { sendBuffer(ServerCnxnFactory.closeConn); } }
NettyServerCnxn extends ServerCnxn { @Override public void sendCloseSession() { sendBuffer(ServerCnxnFactory.closeConn); } NettyServerCnxn(Channel channel, ZooKeeperServer zks, NettyServerCnxnFactory factory); }
NettyServerCnxn extends ServerCnxn { @Override public void sendCloseSession() { sendBuffer(ServerCnxnFactory.closeConn); } NettyServerCnxn(Channel channel, ZooKeeperServer zks, NettyServerCnxnFactory factory); @Override void close(); @Override long getSessionId(); @Override int getSessionTimeout(); @Override void process(WatchedEvent event); @Override void sendResponse(ReplyHeader h, Record r, String tag); @Override void setSessionId(long sessionId); @Override void sendBuffer(ByteBuffer sendBuffer); @Override void disableRecv(); @Override void enableRecv(); @Override long getOutstandingRequests(); @Override void setSessionTimeout(int sessionTimeout); @Override int getInterestOps(); @Override InetSocketAddress getRemoteSocketAddress(); @Override void sendCloseSession(); @Override boolean isSecure(); @Override Certificate[] getClientCertificateChain(); @Override void setClientCertificateChain(Certificate[] chain); }
NettyServerCnxn extends ServerCnxn { @Override public void sendCloseSession() { sendBuffer(ServerCnxnFactory.closeConn); } NettyServerCnxn(Channel channel, ZooKeeperServer zks, NettyServerCnxnFactory factory); @Override void close(); @Override long getSessionId(); @Override int getSessionTimeout(); @Override void process(WatchedEvent event); @Override void sendResponse(ReplyHeader h, Record r, String tag); @Override void setSessionId(long sessionId); @Override void sendBuffer(ByteBuffer sendBuffer); @Override void disableRecv(); @Override void enableRecv(); @Override long getOutstandingRequests(); @Override void setSessionTimeout(int sessionTimeout); @Override int getInterestOps(); @Override InetSocketAddress getRemoteSocketAddress(); @Override void sendCloseSession(); @Override boolean isSecure(); @Override Certificate[] getClientCertificateChain(); @Override void setClientCertificateChain(Certificate[] chain);  }