package com.excavator.rpc.core.server.impl;

import com.excavator.rpc.core.protocol.RpcDecoder;
import com.excavator.rpc.core.protocol.RpcEncoder;
import com.excavator.rpc.core.server.RpcServerHandler;
import com.excavator.rpc.core.server.Server;
import com.excavator.rpc.core.utils.NetUtils;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * Created by cmonkey on 3/28/17.
 */
public class ServerImpl implements Server{
    private static final Logger logger = LoggerFactory.getLogger(ServerImpl.class);
    private String localIp;
    private int port;
    private boolean started = false;
    private Channel channel;
    private Object serviceImpl;
    private String serviceName;
    private String zkConn;
    private String serviceRegisterPath;
    private EventLoopGroup bossGroup = new NioEventLoopGroup();
    private EventLoopGroup workerGroup = new NioEventLoopGroup();
    private CuratorFramework curatorFramework;
    private static final String ZK_DATA_PATH = "";
    public ServerImpl(int prot, Object serviceImpl, String serviceName){
        this.port = port;
        this.serviceImpl = serviceImpl;
        this.serviceName = serviceName;
    }

    public ServerImpl(int port, Object serviceImpl, String serviceName, String zkConn){
        this.port  = port;
        this.serviceImpl = serviceImpl;
        this.serviceName = serviceName;
        this.zkConn = zkConn;
    }
    @Override
    public void start() {
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup,workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                .addLast(new LoggingHandler(LogLevel.INFO))
                                .addLast(new RpcDecoder(10 * 1024 * 1024))
                                .addLast(new RpcEncoder())
                                .addLast(new RpcServerHandler(serviceImpl));
                    }
                });

        // 调用bind 等待客户端链接
        try {
            ChannelFuture future = serverBootstrap.bind(port).sync();
            registerService();
            logger.info("Server started at {}",port);
            started = true;
            this.channel = future.channel();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    private String getZkConn(){
        return "";
    }

    private void registerService(){
        zkConn = getZkConn();
        localIp = NetUtils.getLocalIp();
        String serviceIp = localIp + ":" + port;

        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(zkConn,
                new ExponentialBackoffRetry(1000, 3));
        curatorFramework.start();

        String serviceBasePath = ZK_DATA_PATH + serviceName;

        try {
            curatorFramework.create()
                    .creatingParentsIfNeeded()
                    .forPath(serviceBasePath);
        } catch (Exception e) {
            e.printStackTrace();
            if(e.getMessage().contains("NodeExist")){
                logger.error("this path service has already exist");
            }else{
                logger.error("Create Path error = {}", e);
                throw new RuntimeException("Register error");
            }
        }

        boolean registerSuccess = false;

        while(!registerSuccess){

            try {
                curatorFramework.create()
                        .withMode(CreateMode.EPHEMERAL)
                        .forPath(serviceBasePath + "/" + serviceIp);
                registerSuccess = true;
            } catch (Exception e) {
                e.printStackTrace();
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                logger.info("retry register zk = {}", e.getMessage());

                try {
                    curatorFramework.delete().forPath(serviceBasePath + "/" + serviceIp);
                } catch (Exception e1) {
                    e1.printStackTrace();
                }
            }
        }

    }

    @Override
    public void shutdown() {
        logger.info("shutting down server = {}", serviceName);
        unRegister();

        if(null != curatorFramework){
            curatorFramework.close();
        }

        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }

    private void unRegister(){
        logger.info("unRegister zookeeper");

        try {
            curatorFramework.delete().forPath(ZK_DATA_PATH + serviceName + "/" + localIp);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public String getLocalIp() {
        return localIp;
    }

    public void setLocalIp(String localIp) {
        this.localIp = localIp;
    }

    public Channel getChannel() {
        return channel;
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
    }

    public void setZkConn(String zkConn) {
        this.zkConn = zkConn;
    }
}

@@ -1036,16 +1036,83 @@
 eUnit;%0A%0A
+import static com.excavator.rpc.core.utils.Constant.ZK_DATA_PATH;%0A%0A
 /**%0A * C
@@ -1700,59 +1700,8 @@
 rk;%0A
-    private static final String ZK_DATA_PATH = %22%22;%0A
     
@@ -3268,67 +3268,8 @@
   %7D%0A
-%0A    private String getZkConn()%7B%0A        return %22%22;%0A    %7D%0A%0A
     
@@ -5975,11 +5975,72 @@
 n;%0A    %7D
+%0A%0A    public String getZkConn()%7B%0A        return zkConn;%0A    %7D
 %0A%7D%0A

change get zkconn