public class HelloServer {
    public static void main(String[] args) {
        // 1. 启动器，负责组装 netty 组件，启动服务器
        new ServerBootstrap()
            // 2. ParentEventLoop、ChildEventLoop(Selector,Thread) 相当于 多线程，Parent处理连接事件，Child处理读写事件
            .group(new NioEventLoopGroup())
            // 3. 选择服务器通道 ServerSocketChannel 的实现
            .channel(NioServerSocketChannel.class)
            // 4. Child 处理器,负责处理对应事件
            .childHandler(new ChannelInitializer<NioSocketChannel>() {
                // 5. 初始化 channel
                @Override
                protected void initChannel(NioSocketChannel channel) throws Exception {
                    // 6. 添加具体的处理器
                    channel.pipeline().addLast(new StringDecoder()); // 字符串解码器
                    channel.pipeline().addLast(new ChannelInboundHandlerAdapter(){ // 自定义 handler 处理上一个处理器处理后的结果
                        // 处理通道读事件
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            super.channelRead(ctx, msg);
                            System.out.println(msg); // 打印传输内容
                        }
                    });
                }
            })
            // 7.设置监听端口
            .bind(8080);
    }
}

public class HelloClient {
    public static void main(String[] args) throws InterruptedException {
        // 1. 启动类
        new Bootstrap()
                // 2. 添加 EventLoop
                .group(new NioEventLoopGroup())
                // 3. 选择客户端 channel 实现
                .channel(NioSocketChannel.class)
                // 4. 添加处理器
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    // 连接建立后初始化
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline().addLast(new StringEncoder());
                    }
                })
                // 5. 设置连接地址端口
                .connect(new InetSocketAddress("localhost",8080))
                .sync() // 阻塞方法，直到连接建立
                .channel() // 代表连接对象
                // 6. 向服务器发送数据
                .writeAndFlush("hello,world"); // 写入数据并清空缓冲区
    }
}

// 内部创建了两个 EventLoop, 每个 EventLoop 维护一个线程
DefaultEventLoopGroup group = new DefaultEventLoopGroup(2);
System.out.println(group.next());
System.out.println(group.next());
System.out.println(group.next());

DefaultEventLoopGroup group = new DefaultEventLoopGroup(2);
for (EventExecutor eventLoop : group) {
    System.out.println(eventLoop);
}

public class EventLoopServer {

    private static final Logger log = LoggerFactory.getLogger(EventLoopServer.class);

    public static void main(String[] args) throws InterruptedException {
        new ServerBootstrap()
                // 参数1:boss 只负责 ServerSocketChannel 的 accept 事件,不需要指定线程数，因为只有一个服务端与之进行绑定
                // 参数2:worker 只负责 SocketChannel 的 读写事件
                .group(new NioEventLoopGroup(),new NioEventLoopGroup(2))
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf byteBuf = (ByteBuf) msg;
                                log.debug(byteBuf.toString(StandardCharsets.UTF_8));
                            }
                        });
                    }
                })
                .bind(8080)
                .sync();
    }
}

public class EventLoopClient {
    public static void main(String[] args) throws InterruptedException {
        // 1. 启动类
        Channel channel = new Bootstrap()
                // 2. 添加 EventLoop
                .group(new NioEventLoopGroup())
                // 3. 选择客户端 channel 实现
                .channel(NioSocketChannel.class)
                // 4. 添加处理器
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    // 连接建立后初始化
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline().addLast(new StringEncoder());
                    }
                })
                // 5. 设置连接地址端口
                .connect(new InetSocketAddress("localhost", 8080))
                .sync() // 阻塞方法，直到连接建立
                .channel();// 代表连接对象

        System.out.println(channel);
        System.out.println();
    }
}

public class EventLoopServer {

    private static final Logger log = LoggerFactory.getLogger(EventLoopServer.class);

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

        // 创建一个独立的 EventLoopGroup ，用于处理一些耗时的任务，避免耗时过长，阻塞 io 线程，导致业务停滞
        DefaultEventLoopGroup group = new DefaultEventLoopGroup();

        new ServerBootstrap()
                // 参数1:boss 只负责 ServerSocketChannel 的 accept 事件,不需要指定线程数，因为只有一个服务端与之进行绑定
                // 参数2:worker 只负责 SocketChannel 的 读写事件
                .group(new NioEventLoopGroup(),new NioEventLoopGroup(2))
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline().addLast("nio handler",new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf byteBuf = (ByteBuf) msg;
                                log.debug(byteBuf.toString(StandardCharsets.UTF_8));
                                ctx.fireChannelRead(msg);   // 将消息传递给下一个 handler
                            }
                        }).addLast(group,"default handler",new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf byteBuf = (ByteBuf) msg;
                                log.debug(byteBuf.toString(StandardCharsets.UTF_8));
                            }
                        });
                    }
                })
                .bind(8080)
                .sync();
    }
}

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
    // 下一个 handler 的事件循环是否与当前的事件循环是同一个线程
    EventExecutor executor = next.executor();
    
    // 是，直接调用
    if (executor.inEventLoop()) {
        next.invokeChannelRead(m);
    } 
    // 不是，将要执行的代码作为任务提交给下一个事件循环处理（换人）
    else {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                next.invokeChannelRead(m);
            }
        });
    }
}

public class TestEventLoop {

    private static final Logger log = LoggerFactory.getLogger(TestEventLoop.class);

    public static void main(String[] args) {
        // 1. 创建事件循环组
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(); // io 事件、普通任务、定时任务
        // 默认线程数 = 核心数 * 2
        // DefaultEventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup(); // 普通任务、定时任务
        // 2. 获取下一个事件循环对象
        System.out.println(eventLoopGroup.next());
        System.out.println(eventLoopGroup.next());
        System.out.println(eventLoopGroup.next());

        // 3. 执行普通任务，用于实现异步任务
        eventLoopGroup.next().execute(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("execute ok");
        });

        // 4. 执行定时任务，用于实现心跳检测
        eventLoopGroup.next().scheduleAtFixedRate(() -> {
           log.debug("ok");
        },0,1, TimeUnit.SECONDS);

        // 参数1: Runnable 任务对象
        // 参数2 l = 0 : 初始延迟时间， 0 表示立刻执行
        // 参数3 l1 = 1 : 间隔时间
        // 参数4 时间单位
        // 立刻开始 每间隔 1s 执行一次 debug 输出 ok 的操作
        
        log.debug("main thread..");
    }
}

public class EventLoopClient {
    public static void main(String[] args) throws InterruptedException {
        // 1. 启动类
        ChannelFuture channelFuture = new Bootstrap()
            // 2. 添加 EventLoop
            .group(new NioEventLoopGroup())
            // 3. 选择客户端 channel 实现
            .channel(NioSocketChannel.class)
            // 4. 添加处理器
            .handler(new ChannelInitializer<NioSocketChannel>() {
                // 连接建立后初始化
                @Override
                protected void initChannel(NioSocketChannel channel) throws Exception {
                    channel.pipeline().addLast(new StringEncoder());
                }
            })
            // 5. 连接服务器地址端口
            // 异步非阻塞，main 线程发起调用，真正执行 connect 的是 nio 线程
            .connect(new InetSocketAddress("localhost", 8080)); // 1s 后才建立好连接

        channelFuture.sync(); // 阻塞方法，直到连接建立
        // 无阻塞向下执行获取 channel ，拿到的是 没有建立好连接的 channel ，无法进行读写操作
        Channel channel = channelFuture.channel();// 代表连接对象

        channel.writeAndFlush("hello,world");
    }
}

// 5.1 使用 sync 方法同步处理结果，main 线程阻塞等待处理结果
channelFuture.sync(); // 阻塞方法，直到连接建立

// 无阻塞向下执行获取 channel ，拿到的是 没有建立好连接的 channel ，无法进行读写操作
Channel channel = channelFuture.channel();// 代表连接对象
log.debug("{}",channel);
channel.writeAndFlush("hello,world");

// 5.2 使用 addListener(回调对象) 异步处理结果，不再是 main 线程处理
channelFuture.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        Channel channel = channelFuture.channel();
        log.debug("{}",channel);
        channel.writeAndFlush("hello,world");
    }
});

public class CloseFutureClient {

    private static final Logger log = LoggerFactory.getLogger(CloseFutureClient.class);

    public static void main(String[] args) throws InterruptedException {
        ChannelFuture channelFuture = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        channel.pipeline().addLast(new StringEncoder());
                    }
                })
                .connect(new InetSocketAddress("localhost", 8080));

        Channel channel = channelFuture.sync().channel();

        new Thread(()->{
            Scanner sc = new Scanner(System.in);
            while (true) {
                String s = sc.nextLine();

                if ("q".equals(s)) {
                    channel.close(); // close 异步操作，交给其他线程处理，可能存在延迟关闭
                    // log.debug("处理关闭之后的操作"); // 由于不是阻塞方法，当前一条语句执行完之后，立刻打印日志，所以不能在这里处理关闭后的操作
                    break;
                }

                channel.writeAndFlush(s);
            }
        },"input").start();

        // 获取 CloseFuture 对象
        ChannelFuture closeFuture = channel.closeFuture();
        log.debug("waiting close ...");
        // 1. 同步处理关闭
//        closeFuture.sync();
//        log.debug("处理关闭之后的操作");
//        eventLoopGroup.shutdownGracefully(); 

        // 2. 异步处理关闭
        closeFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                log.debug("处理关闭之后的操作");
                eventLoopGroup.shutdownGracefully(); // 拒绝之后的所有请求，等待正在执行的请求完成之后关闭线程
            }
        });
    }
}

public class TestJdkFuture {

    private static final Logger log = LoggerFactory.getLogger(TestJdkFuture.class);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建线程池
        ExecutorService service = Executors.newFixedThreadPool(2);
        // 2. 提交任务
        Future<Integer> future = service.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                log.debug("执行任务...");
                Thread.sleep(1000);
                return 50;
            }
        });

        log.debug("等待结果...");
        // 3. main 线程通过 future.get() 同步方法,阻塞等待,获取结果
        log.debug("结果为{}",future.get());
    }
}

public class TestNettyFuture {

    private static final Logger log = LoggerFactory.getLogger(TestNettyFuture.class);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();

        EventLoop eventLoop = group.next();

        Future<Integer> future = eventLoop.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                log.debug("执行任务...");
                Thread.sleep(1000);
                return 50;
            }
        });

        log.debug("等待结果...");
        // 3. main 线程通过 future.get() 同步方法,阻塞等待,获取结果
        log.debug("结果为{}",future.get());

        // 4. main 线程通过 future.addListener 回调对象，让执行线程处理结果
        future.addListener(new GenericFutureListener<Future<? super Integer>>() {
            @Override
            public void operationComplete(Future<? super Integer> future) throws Exception {
                log.debug("接收结果:{}",future.getNow());
            }
        });
    }
}

public class TestNettyPromise {

    private static final Logger log = LoggerFactory.getLogger(TestNettyPromise.class);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建 EventLoop
        EventLoop eventLoop = new NioEventLoopGroup().next();
        // 2. 主动创建 promise，结果的容器，相当于线程之间传输数据的通道？一种通信方式？
        DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);

        new Thread(()->{
            // 3. 任意一个线程执行完成后向 promise 填充结果
            log.debug("执行任务...");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            promise.setSuccess(50);
        }).start();

        log.debug("等待结果...");

        log.debug("结果为{}",promise.getNow()); // 此时线程还未执行完成,结果为null

        // 4. main 线程使用 promise.get() 同步接收结果
        log.debug("结果为{}",promise.get());
    }
}

public class TestNettyPromise {

    private static final Logger log = LoggerFactory.getLogger(TestNettyPromise.class);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建 EventLoop
        EventLoop eventLoop = new NioEventLoopGroup().next();
        // 2. 主动创建 promise，结果的容器，相当于线程之间传输数据的通道？一种通信方式？
        DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);

        new Thread(()->{
            // 3. 任意一个线程执行完成后向 promise 填充结果
            log.debug("执行任务...");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            promise.setSuccess(50);
        }).start();

        log.debug("等待结果...");

        // 4. main 线程通过 promise.addListener 回调对象，让执行线程处理结果
        promise.addListener(new GenericFutureListener<Future<? super Integer>>() {
            @Override
            public void operationComplete(Future<? super Integer> future) throws Exception {
                // 这里的 future 就是上面的 promise
                log.debug("接收结果:{}",future.getNow());
            }
        });
    }
}

public class TestNettyPromise {

    private static final Logger log = LoggerFactory.getLogger(TestNettyPromise.class);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建 EventLoop
        EventLoop eventLoop = new NioEventLoopGroup().next();
        // 2. 主动创建 promise，结果的容器，相当于线程之间传输数据的通道？一种通信方式？
        DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);

        new Thread(()->{
            // 3. 任意一个线程执行完成后向 promise 填充结果
            log.debug("执行任务...");
            try {
                int i = 1/0;
                Thread.sleep(1000);
                promise.setSuccess(50);
            } catch (Exception e) {
                e.printStackTrace();
                promise.setFailure(e);
            }
        }).start();

        log.debug("等待结果...");

        log.debug("结果为{}",promise.getNow()); // 此时线程还未执行完成,结果为null

        // 4. main 线程使用 promise.get() 同步接收结果
        log.debug("结果为{}",promise.get()); // sync() 也会出现异常，只是 get 会再用 ExecutionException 包一层异常
    }
}

public class TestNettyPromise {

    private static final Logger log = LoggerFactory.getLogger(TestNettyPromise.class);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建 EventLoop
        EventLoop eventLoop = new NioEventLoopGroup().next();
        // 2. 主动创建 promise，结果的容器，相当于线程之间传输数据的通道？一种通信方式？
        DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);

        new Thread(()->{
            // 3. 任意一个线程执行完成后向 promise 填充结果
            log.debug("执行任务...");
            try {
                int i = 1/0;
                Thread.sleep(1000);
                promise.setSuccess(50);
            } catch (Exception e) {
                e.printStackTrace();
                promise.setFailure(e);
            }
        }).start();

        log.debug("等待结果...");

        log.debug("结果为{}",promise.getNow()); // 此时线程还未执行完成,结果为null

        promise.await();  // 与 sync 和 get 区别在于，不会抛异常

        // 4. 判断 promise 任务是否成功,如果是就返回结果，否则返回错误信息
        log.debug("result {}", (promise.isSuccess() ? promise.getNow() : promise.cause()).toString());
    }
}

public class TestNettyPromise {

    private static final Logger log = LoggerFactory.getLogger(TestNettyPromise.class);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建 EventLoop
        EventLoop eventLoop = new NioEventLoopGroup().next();
        // 2. 主动创建 promise，结果的容器，相当于线程之间传输数据的通道？一种通信方式？
        DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);

        new Thread(()->{
            // 3. 任意一个线程执行完成后向 promise 填充结果
            log.debug("执行任务...");
            try {
                int i = 1/0;
                Thread.sleep(1000);
                promise.setSuccess(50);
            } catch (Exception e) {
                e.printStackTrace();
                promise.setFailure(e);
            }
        }).start();

        log.debug("等待结果...");

        log.debug("结果为{}",promise.getNow()); // 此时线程还未执行完成,结果为null


        // 4. 判断 promise 任务是否成功,如果是就返回结果，否则返回错误信息
        promise.addListener(new GenericFutureListener<Future<? super Integer>>() {
            @Override
            public void operationComplete(Future<? super Integer> future) throws Exception {
                log.debug("result {}", (promise.isSuccess() ? promise.getNow() : promise.cause()).toString());
            }
        });
    }
}

public class TestNettyPromise {

    private static final Logger log = LoggerFactory.getLogger(TestNettyPromise.class);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        DefaultEventLoop eventExecutors = new DefaultEventLoop();
        DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);

        eventExecutors.submit(()->{
            System.out.println("1");
            try {
                promise.await();
                // 注意不能仅捕获 InterruptedException 异常
                // 否则 死锁检查抛出的 BlockingOperationException 会继续向上传播
                // 而提交的任务会被包装为 PromiseTask，它的 run 方法中会 catch 所有异常然后设置为 Promise 的失败结果而不会抛出
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("2");
        });
        eventExecutors.submit(()->{
            System.out.println("3");
            try {
                promise.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("4");
        });
    }
}

public class TestPipeline {

    private static final Logger log = LoggerFactory.getLogger(TestPipeline.class);

    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        // 1. 通过 channel 拿到 pipeline
                        ChannelPipeline pipeline = channel.pipeline();
                        // 2. 添加处理器,自带head,tail处理器 head -> h1 -> h2 -> h3 -> h4 -> h5 -> h6 -> tail,双向链表
                        pipeline.addLast("h1",new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                log.debug("1");
                                ByteBuf byteBuf = (ByteBuf) msg;
                                String name = byteBuf.toString(StandardCharsets.UTF_8); // 获取 name
                                super.channelRead(ctx, name); // 1 传递给下一个 handler
                            }
                        });
                        pipeline.addLast("h2",new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                log.debug("2");
                                Student student = new Student(msg.toString()); // 构造 student
                                super.channelRead(ctx, student); // 2
                            }
                        });
                        pipeline.addLast("h4", new ChannelOutboundHandlerAdapter(){
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                log.debug("4");
                                super.write(ctx, msg, promise); // 4
                            }
                        });
                        pipeline.addLast("h3",new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                log.debug("3");
                                log.debug("class: {} result: {}",msg.getClass(),msg);
                                // super.channelRead(ctx, msg); // 3
                                ctx.writeAndFlush(ctx.alloc().buffer().writeBytes("hello,world".getBytes())); // 从当前处理器向前触发
                                // channel.writeAndFlush(ctx.alloc().buffer().writeBytes("hello,world".getBytes())); // 跳到 tail 处理器向前触发
                                // 出站处理器需要写出才能执行
                            }
                        });
                        pipeline.addLast("h5", new ChannelOutboundHandlerAdapter(){
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                log.debug("5");
                                super.write(ctx, msg, promise); // 5
                            }
                        });
                        pipeline.addLast("h6", new ChannelOutboundHandlerAdapter(){
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                log.debug("6");
                                super.write(ctx, msg, promise); // 6
                            }
                        });
                    }
                })
                .bind(8080);
    }

    static class Student {
        private String name;

        public Student() {
        }

        public Student(String name) {
            this.name = name;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        @Override
        public String toString() {
            return "Student{" +
                    "name='" + name + '\'' +
                    '}';
        }
    }
}

public class TestByteBuf {

    private static final Logger log = LoggerFactory.getLogger(TestByteBuf.class);

    public static void main(String[] args) {

        ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer(); // 默认容量 256,动态扩容
        log(byteBuf);
        StringBuilder sb = new StringBuilder();

        for (int i = 0; i < 32; i++) {
            sb.append("a");
        }

        byteBuf.writeBytes(sb.toString().getBytes());

        log(byteBuf);
    }

    private static void log(ByteBuf buffer) {
        int length = buffer.readableBytes();
        int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
        StringBuilder buf = new StringBuilder(rows * 80 * 2)
                .append("read index:").append(buffer.readerIndex())
                .append(" write index:").append(buffer.writerIndex())
                .append(" capacity:").append(buffer.capacity())
                .append(NEWLINE);
        appendPrettyHexDump(buf, buffer);
        log.debug("{}",buf);
    }
}

ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10);

ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);

-Dio.netty.allocator.type={unpooled|pooled}

buffer.writeBytes(new byte[]{1, 2, 3, 4});
log(buffer);

buffer.writeInt(5);
log(buffer);

buffer.writeInt(6);
log(buffer);

System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
log(buffer);

buffer.markReaderIndex();
System.out.println(buffer.readInt());
log(buffer);

buffer.resetReaderIndex();
log(buffer);

ByteBuf buf = ...
try {
    ...
} finally {
    buf.release();
}

// 实现 ChannelInboundHandler 用于处理入站最后的收尾工作，引用计数法回收内存
final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler

public void channelRead(ChannelHandlerContext ctx, Object msg) {
    DefaultChannelPipeline.this.onUnhandledInboundMessage(ctx, msg);
}

// io.netty.channel.DefaultChannelPipeline#onUnhandledInboundMessage(java.lang.Object)
protected void onUnhandledInboundMessage(Object msg) {
    try {
        logger.debug(
            "Discarded inbound message {} that reached at the tail of the pipeline. " +
            "Please check your pipeline configuration.", msg);
    } finally {
        ReferenceCountUtil.release(msg);
    }
}

// io.netty.util.ReferenceCountUtil#release(java.lang.Object)
public static boolean release(Object msg) {
    if (msg instanceof ReferenceCounted) {
        return ((ReferenceCounted) msg).release();
    }
    return false;
}

// 实现 ChannelInboundHandler 用于处理入站开始工作
// 实现 ChannelOutboundHandler 用于处理出站最后的收尾工作，引用计数法回收内存
final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler

ByteBuf origin = ByteBufAllocator.DEFAULT.buffer(10);
origin.writeBytes(new byte[]{1, 2, 3, 4});
origin.readByte();
System.out.println(ByteBufUtil.prettyHexDump(origin));

ByteBuf slice = origin.slice();
System.out.println(ByteBufUtil.prettyHexDump(slice));
// slice.writeByte(5); 如果执行，会报 IndexOutOfBoundsException 异常

origin.readByte();
System.out.println(ByteBufUtil.prettyHexDump(origin));

slice.setByte(2, 5);
System.out.println(ByteBufUtil.prettyHexDump(slice));

public class TestSlice {

    public static void main(String[] args) {
        ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer(10);
        byteBuf.writeBytes(new byte[]{'a','b','c','d','e','f','g','h','i','j'});
        log(byteBuf);

        // 在切片过程中没有发生数据复制
        ByteBuf slice1 = byteBuf.slice(0, 5);
        ByteBuf slice2 = byteBuf.slice(5, 5);

        log(slice1);
        log(slice2);

        System.out.println("==============================");
        slice1.setByte(0,'b');
        log(byteBuf);
        log(slice1);
    }
}

ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);
buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});
ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);
buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});
System.out.println(ByteBufUtil.prettyHexDump(buf1));
System.out.println(ByteBufUtil.prettyHexDump(buf2));

ByteBuf buf3 = ByteBufAllocator.DEFAULT
    .buffer(buf1.readableBytes()+buf2.readableBytes());
buf3.writeBytes(buf1);
buf3.writeBytes(buf2);
System.out.println(ByteBufUtil.prettyHexDump(buf3));

CompositeByteBuf buf3 = ByteBufAllocator.DEFAULT.compositeBuffer();
// true 表示增加新的 ByteBuf 自动递增 write index, 否则 write index 会始终为 0
buf3.addComponents(true, buf1, buf2);

ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);
buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});
ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);
buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});

// 当包装 ByteBuf 个数超过一个时, 底层使用了 CompositeByteBuf
ByteBuf buf3 = Unpooled.wrappedBuffer(buf1, buf2);
System.out.println(ByteBufUtil.prettyHexDump(buf3));

ByteBuf buf4 = Unpooled.wrappedBuffer(new byte[]{1, 2, 3}, new byte[]{4, 5, 6});
System.out.println(buf4.getClass());
System.out.println(ByteBufUtil.prettyHexDump(buf4));

new ServerBootstrap()
    .group(new NioEventLoopGroup())
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<NioSocketChannel>() {
        @Override
        protected void initChannel(NioSocketChannel ch) {
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    ByteBuf buffer = (ByteBuf) msg;
                    System.out.println(buffer.toString(Charset.defaultCharset()));

                    // 建议使用 ctx.alloc() 创建 ByteBuf
                    ByteBuf response = ctx.alloc().buffer();
                    response.writeBytes(buffer);
                    ctx.writeAndFlush(response);

                    // 思考：需要释放 buffer 吗
                    // 思考：需要释放 response 吗
                }
            });
        }
    }).bind(8080);

NioEventLoopGroup group = new NioEventLoopGroup();
Channel channel = new Bootstrap()
    .group(group)
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<NioSocketChannel>() {
        @Override
        protected void initChannel(NioSocketChannel ch) throws Exception {
            ch.pipeline().addLast(new StringEncoder());
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    ByteBuf buffer = (ByteBuf) msg;
                    System.out.println(buffer.toString(Charset.defaultCharset()));

                    // 思考：需要释放 buffer 吗
                }
            });
        }
    }).connect("127.0.0.1", 8080).sync().channel();

channel.closeFuture().addListener(future -> {
    group.shutdownGracefully();
});

new Thread(() -> {
    Scanner scanner = new Scanner(System.in);
    while (true) {
        String line = scanner.nextLine();
        if ("q".equals(line)) {
            channel.close();
            break;
        }
        channel.writeAndFlush(line);
    }
}).start();

public class TestServer {
    public static void main(String[] args) throws IOException {
        ServerSocket ss = new ServerSocket(8888);
        Socket s = ss.accept();

        new Thread(() -> {
            try {
                BufferedReader reader = new BufferedReader(new InputStreamReader(s.getInputStream()));
                while (true) {
                    System.out.println(reader.readLine());
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();

        new Thread(() -> {
            try {
                BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(s.getOutputStream()));
                // 例如在这个位置加入 thread 级别断点，可以发现即使不写入数据，也不妨碍前面线程读取客户端数据
                for (int i = 0; i < 100; i++) {
                    writer.write(String.valueOf(i));
                    writer.newLine();
                    writer.flush();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

public class TestClient {
    public static void main(String[] args) throws IOException {
        Socket s = new Socket("localhost", 8888);

        new Thread(() -> {
            try {
                BufferedReader reader = new BufferedReader(new InputStreamReader(s.getInputStream()));
                while (true) {
                    System.out.println(reader.readLine());
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();

        new Thread(() -> {
            try {
                BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(s.getOutputStream()));
                for (int i = 0; i < 100; i++) {
                    writer.write(String.valueOf(i));
                    writer.newLine();
                    writer.flush();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();
    }
}