package com.njq.study.socket.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.concurrent.TimeUnit;

/**
 * @author: nijiaqi
 * @date: 2019/12/18
 */
public class TimeServer {
    public void bind(int port) {
        try {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .childHandler(new ChildChannelHandler());

                ChannelFuture f = b.bind(port).sync();
                f.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
//            socketChannel.pipeline().addLast(new TimeServerHandler());
            ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
            socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
            socketChannel.pipeline().addLast(new StringDecoder());
            socketChannel.pipeline().addLast(new TimeServerHandler());
        }
    }

    public static void main(String[] args) throws Exception {
        int port = 56789;
        try {
            new Thread(() -> new TimeServer().bind(port)).start();
//            System.out.println(123123123123L);
//            TimeUnit.SECONDS.sleep(2);
//            BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
//            String str;
//            System.out.println("Enter lines of text.");
//            System.out.println("Enter 'exit' to quit.");
//            do {
//                // 从控制台读取一行数据，返回值字符串
//                str = br.readLine();
//                System.out.println("--------------");
//                System.out.println(str);
//                ByteBuf echo1 = Unpooled.copiedBuffer(str.getBytes());
//                TimeServerHandler.map.get("123").writeAndFlush(echo1);
//            } while (!str.equals("exit"));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
        }
    }

}

package com.njq.study.socket.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.Map;


public class TimeServerHandler extends ChannelInboundHandlerAdapter {
    public static Map<String, ChannelHandlerContext> map = new HashMap();

    int counter = 0;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        map.put("123", ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//        ByteBuf buf = (ByteBuf) msg;
//        byte[] req = new byte[buf.readableBytes()];
//        buf.readBytes(req);
//        String body = new String(req, "UTF-8");
//        String currentTime = "".equals(body) ? String.valueOf(new Date().getTime()) : "BAD ORDER\r\n";
//        System.out.println("receive:"+body+(++counter));
//        String currentTime = "ERROR";
//        if (body.equals("abc\r\n")) {
//            currentTime = String.valueOf(System.currentTimeMillis());
//        }
//        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
//        ctx.write(resp);
//        ctx.flush();

        String body = (String)msg;
        System.out.println("The time server receive order:" + (++counter) + body);
        body += "$_";
        System.out.println("write str:"+body);
        ByteBuf echo =Unpooled.copiedBuffer(body.getBytes());
        ctx.writeAndFlush(echo);
//
//
//
//        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
//        String str;
//        System.out.println("Enter lines of text.");
//        System.out.println("Enter 'exit' to quit.");
//        do {
//            // 从控制台读取一行数据，返回值字符串
//            str = br.readLine();
//            System.out.println(str);
//            ByteBuf echo1 =Unpooled.copiedBuffer(str.getBytes());
//            ctx.writeAndFlush(echo1);
//        } while (!str.equals("exit"));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

package com.njq.study.socket.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

public class TimerClient {
    public void connect(int port, String host) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
//                            socketChannel.pipeline().addLast(new TimerClientHandler());
                            ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
                            socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
                            socketChannel.pipeline().addLast(new StringDecoder());
                            socketChannel.pipeline().addLast(new TimerClientHandler());
                        }
                    });
            ChannelFuture f = b.connect(host, port).sync();

            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) {
        int port = 56789;
        try {
            new TimerClient().connect(port, "localhost");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
        }
    }
}

package com.njq.study.socket.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class TimerClientHandler extends ChannelInboundHandlerAdapter {
    private final ByteBuf firstMessage;

    public TimerClientHandler() {
        byte[] req = "abc\r\n".getBytes();
        this.firstMessage = Unpooled.buffer(req.length);
        firstMessage.writeBytes(req);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
//        ctx.writeAndFlush(firstMessage);
        String echoReq = "Hi,welcom to netty.$_";
        for (int i = 0; i < 20; i++) {
            if(i == 19){
                echoReq = echoReq.replace("_", "+");
            }
            ctx.writeAndFlush(Unpooled.copiedBuffer(echoReq.getBytes()));
        }

        Thread.sleep(2000);
        ctx.writeAndFlush(Unpooled.copiedBuffer("nnnnnn.$_".getBytes()));
    }

    int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//        ByteBuf buf = (ByteBuf) msg;
//        byte[] req = new byte[buf.readableBytes()];
//        buf.readBytes(req);
//        String body = new String(req, "UTF-8");
//        System.out.println("now is :" + (++counter) + ":" + body);


        String body = (String)msg;
        System.out.println("now is :" + (++counter) + ":" + body);

//        ByteBuf msg1 = Unpooled.wrappedBuffer("msg hello world".getBytes());
//        ctx.writeAndFlush(msg1);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}