@Override
protected boolean handleProxyProtocol(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (!authenticated) {
        authenticated = authenticate(ctx, msg);
        return false;
    }

    Socks5CommandRequest req = (Socks5CommandRequest) msg;
    assertThat(req.type(), is(Socks5CommandType.CONNECT));

    Socks5CommandResponse res =
            new DefaultSocks5CommandResponse(Socks5CommandStatus.SUCCESS, Socks5AddressType.IPv4);
    intermediaryDestination = SocketUtils.socketAddress(req.dstAddr(), req.dstPort());

    ctx.write(res);

    ctx.pipeline().remove(ENCODER);
    ctx.pipeline().remove(DECODER);

    return true;
}
 

@Ignore
@Test
public void testSubnetQuery() throws Exception {
    EventLoopGroup group = new NioEventLoopGroup(1);
    DnsNameResolver resolver = newResolver(group).build();
    try {
        // Same as:
        // # /.bind-9.9.3-edns/bin/dig @ns1.google.com www.google.es +client=157.88.0.0/24
        Future<List<InetAddress>> future = resolver.resolveAll("www.google.es",
                Collections.<DnsRecord>singleton(
                        // Suggest max payload size of 1024
                        // 157.88.0.0 / 24
                        new DefaultDnsOptEcsRecord(1024, 24,
                                                   SocketUtils.addressByName("157.88.0.0").getAddress())));
        for (InetAddress address: future.syncUninterruptibly().getNow()) {
            System.err.println(address);
        }
    } finally {
        resolver.close();
        group.shutdownGracefully(0, 0, TimeUnit.SECONDS);
    }
}
 

@Override
protected boolean doConnect(final SocketAddress remoteAddress,
        final SocketAddress localAddress) throws Exception {
    doBind(localAddress != null? localAddress : new InetSocketAddress(0));
    boolean success = false;
    try {
        final boolean connected = SocketUtils.connect(javaChannel(), remoteAddress);
        if (!connected) {
            selectionKey().interestOps(
                    selectionKey().interestOps() | OP_CONNECT);
        }
        success = true;
        return connected;
    } finally {
        if (!success) {
            doClose();
        }
    }
}
 

@Override
protected boolean doConnect(final SocketAddress remoteAddress,
                            final SocketAddress localAddress) throws Exception {
    doBind(localAddress != null? localAddress : new InetSocketAddress(0));
    boolean success = false;
    try {
        final boolean connected = SocketUtils.connect(javaChannel(), remoteAddress);
        if (!connected) {
            selectionKey().interestOps(
                    selectionKey().interestOps() | OP_CONNECT);
        }
        success = true;
        return connected;
    } finally {
        if (!success) {
            doClose();
        }
    }
}
 

@Override
protected void doConnect(SocketAddress remoteAddress,
        SocketAddress localAddress) throws Exception {
    if (localAddress != null) {
        SocketUtils.bind(socket, localAddress);
    }

    boolean success = false;
    try {
        SocketUtils.connect(socket, remoteAddress, config().getConnectTimeoutMillis());
        activate(socket.getInputStream(), socket.getOutputStream());
        success = true;
    } catch (SocketTimeoutException e) {
        ConnectTimeoutException cause = new ConnectTimeoutException("connection timed out: " + remoteAddress);
        cause.setStackTrace(e.getStackTrace());
        throw cause;
    } finally {
        if (!success) {
            doClose();
        }
    }
}
 

@Override
protected int doReadMessages(List<Object> buf) throws Exception {
    SocketChannel ch = SocketUtils.accept(javaChannel());

    try {
        if (ch != null) {
            buf.add(new NioSocketChannel(this, ch));
            return 1;
        }
    } catch (Throwable t) {
        logger.warn("Failed to create a new channel from an accepted socket.", t);

        try {
            ch.close();
        } catch (Throwable t2) {
            logger.warn("Failed to close a socket.", t2);
        }
    }

    return 0;
}
 

@Override
    protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
        if (localAddress != null) {
//            调用java底层的socketChannel的bind方法
            doBind0(localAddress);
        }

        boolean success = false;
        try {
//            调用java底层的socketChannel的connect方法
            boolean connected = SocketUtils.connect(javaChannel(), remoteAddress);
            if (!connected) {
//                如果没有连接成功注册OP_CONNECT事件
                selectionKey().interestOps(SelectionKey.OP_CONNECT);
            }
            success = true;
            return connected;
        } finally {
            if (!success) {
//                如果连接失败调用java底层channel close
                doClose();
            }
        }
    }
 

public static void main(String[] args) {
    HydraClient client = new Client.Builder("localhost", 8888, new UdpClientProtocol())
            .useUDP(true)
            .option(ChannelOption.SO_BROADCAST, true)
            .build();

    if (client.isConnected()) {
        session = client.getSession();
        System.out.println("\nClient is online!");
        System.out.printf("Socket address: %s%n", session.getAddress());
    }

    System.out.println(session);

    try {
        session.getChannel().writeAndFlush(new DatagramPacket(Unpooled.copiedBuffer("QOTM?", CharsetUtil.UTF_8),
                SocketUtils.socketAddress("localhost", 8888))).sync();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    // Send something simple
    session.send("This is a String and dealt with as object by Hydra");
}
 

public static void main(String[] args) throws Exception {

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioDatagramChannel.class)
             .option(ChannelOption.SO_BROADCAST, true)
             .handler(new QuoteOfTheMomentClientHandler());

            Channel ch = b.bind(0).sync().channel();

            // Broadcast the QOTM request to port 8080.
            ch.writeAndFlush(new DatagramPacket(
                    Unpooled.copiedBuffer("QOTM?", CharsetUtil.UTF_8),
                    SocketUtils.socketAddress("255.255.255.255", PORT))).sync();

            // QuoteOfTheMomentClientHandler will close the DatagramChannel when a
            // response is received.  If the channel is not closed within 5 seconds,
            // print an error message and quit.
            if (!ch.closeFuture().await(5000)) {
                System.err.println("QOTM request timed out.");
            }
        } finally {
            group.shutdownGracefully();
        }
    }
 

@Test
public void writeQueryTest() throws Exception {
    InetSocketAddress addr = SocketUtils.socketAddress("8.8.8.8", 53);
    EmbeddedChannel embedder = new EmbeddedChannel(new DatagramDnsQueryEncoder());
    List<DnsQuery> queries = new ArrayList<DnsQuery>(5);
    queries.add(new DatagramDnsQuery(null, addr, 1).setRecord(
            DnsSection.QUESTION,
            new DefaultDnsQuestion("1.0.0.127.in-addr.arpa", DnsRecordType.PTR)));
    queries.add(new DatagramDnsQuery(null, addr, 1).setRecord(
            DnsSection.QUESTION,
            new DefaultDnsQuestion("www.example.com", DnsRecordType.A)));
    queries.add(new DatagramDnsQuery(null, addr, 1).setRecord(
            DnsSection.QUESTION,
            new DefaultDnsQuestion("example.com", DnsRecordType.AAAA)));
    queries.add(new DatagramDnsQuery(null, addr, 1).setRecord(
            DnsSection.QUESTION,
            new DefaultDnsQuestion("example.com", DnsRecordType.MX)));
    queries.add(new DatagramDnsQuery(null, addr, 1).setRecord(
            DnsSection.QUESTION,
            new DefaultDnsQuestion("example.com", DnsRecordType.CNAME)));

    for (DnsQuery query: queries) {
        assertThat(query.count(DnsSection.QUESTION), is(1));
        assertThat(query.count(DnsSection.ANSWER), is(0));
        assertThat(query.count(DnsSection.AUTHORITY), is(0));
        assertThat(query.count(DnsSection.ADDITIONAL), is(0));

        embedder.writeOutbound(query);

        DatagramPacket packet = embedder.readOutbound();
        Assert.assertTrue(packet.content().isReadable());
        packet.release();
        Assert.assertNull(embedder.readOutbound());
    }
}
 

public void testShutdownOutput(ServerBootstrap sb) throws Throwable {
    TestHandler h = new TestHandler();
    Socket s = new Socket();
    Channel sc = null;
    try {
        sc = sb.childHandler(h).childOption(ChannelOption.ALLOW_HALF_CLOSURE, true).bind().sync().channel();

        SocketUtils.connect(s, sc.localAddress(), 10000);
        s.getOutputStream().write(1);

        assertEquals(1, (int) h.queue.take());

        assertTrue(h.ch.isOpen());
        assertTrue(h.ch.isActive());
        assertFalse(h.ch.isInputShutdown());
        assertFalse(h.ch.isOutputShutdown());

        s.shutdownOutput();

        h.halfClosure.await();

        assertTrue(h.ch.isOpen());
        assertTrue(h.ch.isActive());
        assertTrue(h.ch.isInputShutdown());
        assertFalse(h.ch.isOutputShutdown());
        assertEquals(1, h.closure.getCount());
        Thread.sleep(100);
        assertEquals(1, h.halfClosureCount.intValue());
    } finally {
        if (sc != null) {
            sc.close();
        }
        s.close();
    }
}
 

public void testShutdownOutputWithoutOption(ServerBootstrap sb) throws Throwable {
    TestHandler h = new TestHandler();
    Socket s = new Socket();
    Channel sc = null;
    try {
        sc = sb.childHandler(h).bind().sync().channel();

        SocketUtils.connect(s, sc.localAddress(), 10000);
        s.getOutputStream().write(1);

        assertEquals(1, (int) h.queue.take());

        assertTrue(h.ch.isOpen());
        assertTrue(h.ch.isActive());
        assertFalse(h.ch.isInputShutdown());
        assertFalse(h.ch.isOutputShutdown());

        s.shutdownOutput();

        h.closure.await();

        assertFalse(h.ch.isOpen());
        assertFalse(h.ch.isActive());
        assertTrue(h.ch.isInputShutdown());
        assertTrue(h.ch.isOutputShutdown());

        assertEquals(1, h.halfClosure.getCount());
        Thread.sleep(100);
        assertEquals(0, h.halfClosureCount.intValue());
    } finally {
        if (sc != null) {
            sc.close();
        }
        s.close();
    }
}
 

public void testNonCachedResolveEmptyHostName(String inetHost) throws Exception {
    DnsNameResolver resolver = newNonCachedResolver(ResolvedAddressTypes.IPV4_ONLY).build();
    try {
        InetAddress addr = resolver.resolve(inetHost).syncUninterruptibly().getNow();
        assertEquals(SocketUtils.addressByName(inetHost), addr);
    } finally {
        resolver.close();
    }
}
 

private static void testNonCachedResolveAllEmptyHostName(String inetHost) throws UnknownHostException {
    DnsNameResolver resolver = newNonCachedResolver(ResolvedAddressTypes.IPV4_ONLY).build();
    try {
        List<InetAddress> addrs = resolver.resolveAll(inetHost).syncUninterruptibly().getNow();
        assertEquals(Arrays.asList(
                SocketUtils.allAddressesByName(inetHost)), addrs);
    } finally {
        resolver.close();
    }
}
 

private static DnsNameResolverBuilder newResolver(EventLoopGroup group) {
    return new DnsNameResolverBuilder(group.next())
            .channelType(NioDatagramChannel.class)
            .nameServerProvider(
                    new SingletonDnsServerAddressStreamProvider(SocketUtils.socketAddress("8.8.8.8", 53)))
            .maxQueriesPerResolve(1)
            .optResourceEnabled(false)
            .ndots(1);
}
 

@Override
protected int doReadMessages(List<Object> buf) throws Exception {
    final SocketChannelUDT channelUDT = (SocketChannelUDT) SocketUtils.accept(javaChannel());
    if (channelUDT == null) {
        return 0;
    } else {
        buf.add(newConnectorChannel(channelUDT));
        return 1;
    }
}
 

public IpSubnetFilterRule(String ipAddress, int cidrPrefix, IpFilterRuleType ruleType) {
    try {
        filterRule = selectFilterRule(SocketUtils.addressByName(ipAddress), cidrPrefix, ruleType);
    } catch (UnknownHostException e) {
        throw new IllegalArgumentException("ipAddress", e);
    }
}
 

private static EmbeddedChannel newEmbeddedInetChannel(final String ipAddress, ChannelHandler... handlers) {
    return new EmbeddedChannel(handlers) {
        @Override
        protected SocketAddress remoteAddress0() {
            return isActive()? SocketUtils.socketAddress(ipAddress, 5421) : null;
        }
    };
}
 

@Override
public InetAddress getInterface() {
    NetworkInterface inf = getNetworkInterface();
    if (inf == null) {
        return null;
    } else {
        Enumeration<InetAddress> addresses = SocketUtils.addressesFromNetworkInterface(inf);
        if (addresses.hasMoreElements()) {
            return addresses.nextElement();
        }
        return null;
    }
}
 

private void doBind0(SocketAddress localAddress) throws Exception {
    if (PlatformDependent.javaVersion() >= 7) {
        SocketUtils.bind(javaChannel(), localAddress);
    } else {
        javaChannel().socket().bind(localAddress);
    }
}
 

private void doBind0(SocketAddress localAddress) throws Exception {
    if (PlatformDependent.javaVersion() >= 7) {
        SocketUtils.bind(javaChannel(), localAddress);
    } else {
        SocketUtils.bind(javaChannel().socket(), localAddress);
    }
}
 

@Test
public void testCmdRequestDecoderIPv6() throws UnknownHostException {
    String[] hosts = {SocksCommonUtils.ipv6toStr(SocketUtils.addressByName("::1").getAddress())};
    int[] ports = {1, 32769, 65535};
    for (SocksCmdType cmdType : SocksCmdType.values()) {
        for (String host : hosts) {
            for (int port : ports) {
                testSocksCmdRequestDecoderWithDifferentParams(cmdType, SocksAddressType.IPv6, host, port);
            }
        }
    }
}
 

@Test
public void testCmdRequestDecoderIPv6() throws UnknownHostException {
    String[] hosts = {
            NetUtil.bytesToIpAddress(SocketUtils.addressByName("::1").getAddress()) };
    int[] ports = {1, 32769, 65535};
    for (Socks5CommandType cmdType: Arrays.asList(Socks5CommandType.BIND,
                                                  Socks5CommandType.CONNECT,
                                                  Socks5CommandType.UDP_ASSOCIATE)) {
        for (String host : hosts) {
            for (int port : ports) {
                test(cmdType, Socks5AddressType.IPv6, host, port);
            }
        }
    }
}
 

@Override
protected void doResolve(String inetHost, Promise<InetAddress> promise) throws Exception {
    try {
        promise.setSuccess(SocketUtils.addressByName(inetHost));
    } catch (UnknownHostException e) {
        promise.setFailure(e);
    }
}
 

@Override
protected void doResolveAll(String inetHost, Promise<List<InetAddress>> promise) throws Exception {
    try {
        promise.setSuccess(Arrays.asList(SocketUtils.allAddressesByName(inetHost)));
    } catch (UnknownHostException e) {
        promise.setFailure(e);
    }
}
 

@Test
public void testEncode() {
    InetSocketAddress recipient = SocketUtils.socketAddress("127.0.0.1", 10000);
    InetSocketAddress sender = SocketUtils.socketAddress("127.0.0.1", 20000);
    assertTrue(channel.writeOutbound(
            new DefaultAddressedEnvelope<String, InetSocketAddress>("netty", recipient, sender)));
    DatagramPacket packet = channel.readOutbound();
    try {
        assertEquals("netty", packet.content().toString(CharsetUtil.UTF_8));
        assertEquals(recipient, packet.recipient());
        assertEquals(sender, packet.sender());
    } finally {
        packet.release();
    }
}
 

@Test
public void testUnmatchedMessageType() {
    InetSocketAddress recipient = SocketUtils.socketAddress("127.0.0.1", 10000);
    InetSocketAddress sender = SocketUtils.socketAddress("127.0.0.1", 20000);
    DefaultAddressedEnvelope<Long, InetSocketAddress> envelope =
            new DefaultAddressedEnvelope<Long, InetSocketAddress>(1L, recipient, sender);
    assertTrue(channel.writeOutbound(envelope));
    DefaultAddressedEnvelope<Long, InetSocketAddress> output = channel.readOutbound();
    try {
        assertSame(envelope, output);
    } finally {
        output.release();
    }
}
 

@Test
public void testDecode() {
    InetSocketAddress recipient = SocketUtils.socketAddress("127.0.0.1", 10000);
    InetSocketAddress sender = SocketUtils.socketAddress("127.0.0.1", 20000);
    ByteBuf content = Unpooled.wrappedBuffer("netty".getBytes(CharsetUtil.UTF_8));
    assertTrue(channel.writeInbound(new DatagramPacket(content, recipient, sender)));
    assertEquals("netty", channel.readInbound());
}
 

private void doBind0(SocketAddress localAddress) throws Exception {
    if (PlatformDependent.javaVersion() >= 7) {
        SocketUtils.bind(javaChannel(), localAddress);
    } else {
        javaChannel().socket().bind(localAddress);
    }
}
 

private void doBind0(SocketAddress localAddress) throws Exception {
    if (PlatformDependent.javaVersion() >= 7) {
        SocketUtils.bind(javaChannel(), localAddress);
    } else {
        javaChannel().socket().bind(localAddress);
    }
}