package chat.server.netty;

import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;

import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import chat.consts.BrokerConstants;
import chat.server.Configer;
import chat.server.moquette.BytesMetricsCollector;
import chat.server.moquette.MessageMetricsCollector;
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.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.Future;

public class NettyAcceptor {

	private static Logger logger;
	
	private static NettyAcceptor instance;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    
    private Class<? extends ServerSocketChannel> channelClass;
    private ChannelFuture httpChannelFuture;
    private ChannelFuture tcpChannelFuture;
    private ChannelFuture sslTCPChannelFuture;
    private ChannelFuture wsChannelFuture;
    
    private MessageMetricsCollector messageMetricsCollector;
    private BytesMetricsCollector bytesMetricsCollector;

	
	static {
		logger = Logger.getLogger(NettyAcceptor.class);
	}
	
	private NettyAcceptor() {
		
	}
	
	public static synchronized NettyAcceptor getInstance() {
		if (instance == null) {
			instance = new NettyAcceptor();
		}
		
		return instance;
	}
	
	public static void start() throws Exception {
		instance = getInstance();
		instance.doStart();
	}
	
	public static void stop() {
		try {
			instance = getInstance();
			instance.doStop();
		}
		catch(Exception e) {
			logger.error(e);
		}
	}
	
    public void doStart() throws Exception {
    	logger.info("Initializing Netty acceptor...");
    	
    	//1.
    	initLocalVariable();
    	
        //2. HTTP 短连接
        initializeHttpTransport();
        
        //3. TCP 长连接
        initializePlainTCPTransport();
        
        //4. SSL 长连接
        initializeSSLTCPTransport();
        
        //5. WebSocket 长连接
        initializeWSSTransport();
        
        logger.info("Netty acceptor initialized");
    }
    
	public void doStop() {
    	logger.info("Closing Netty acceptor...");
    	
        //1.
        closeOneChannel(httpChannelFuture);
        closeOneChannel(tcpChannelFuture);
        closeOneChannel(sslTCPChannelFuture);
        closeOneChannel(wsChannelFuture);
        
        //2.
        closeOneEventLoopGroup("worker group", workerGroup);
        closeOneEventLoopGroup("boss group", bossGroup);
    }

	private void initLocalVariable() {
    	//1. 
        boolean epoll = Configer.getBoolean(BrokerConstants.NETTY_EPOLL_PROPERTY_NAME, false);
        if (epoll) {
            // 由于目前只支持TCP MQTT， 所以bossGroup的线程数配置为1
        	logger.info("Netty is using Epoll");
            bossGroup = new EpollEventLoopGroup(1);
            workerGroup = new EpollEventLoopGroup();
            channelClass = EpollServerSocketChannel.class;
        } 
        else {
        	logger.info("Netty is using NIO");
        	bossGroup = new NioEventLoopGroup(1);
        	workerGroup = new NioEventLoopGroup();
            channelClass = NioServerSocketChannel.class;
        }
        
        //2.
        bytesMetricsCollector = new BytesMetricsCollector();
        messageMetricsCollector = new MessageMetricsCollector();
	}
    
	private void initializeHttpTransport() throws Exception {
		String protocol = "HTTP";	
		Integer port = Configer.getInteger("http_port", null);
		
		if (port == null) {
			logger.info("skip server, Protocol=" + protocol);
			return;
		}
		
		//2. create bootstrap
		logger.info("Initializing server. Protocol=" + protocol);
		NettyHttpChannelInitializer initializer = new NettyHttpChannelInitializer();
		ServerBootstrap bootStrap = new NettyServerBootstrap(this, initializer);
		
		//3. Bind and start to accept incoming connections.
		httpChannelFuture = bootStrap.bind(port);    
		httpChannelFuture.sync().addListener(FIRE_EXCEPTION_ON_FAILURE);
		
		logger.info("server bounded. Protocol=" + protocol + ": " + port);
	}

	private void initializePlainTCPTransport() throws Exception {
		//1. load host and port
		String protocol = "TCP";
		Integer port = Configer.getInteger("port", null);

		if (port == null) {
			logger.info("skip server, Protocol=" + protocol);
			return;
		}
		
		//2. create bootstrap
		NettyTCPChannelInitializer initializer = new NettyTCPChannelInitializer();
        initializer.setMessageMetricsCollector(messageMetricsCollector);
        initializer.setBytesMetricsCollector(bytesMetricsCollector);
        
        ServerBootstrap bootStrap = new NettyServerBootstrap(this, initializer);
        
        //3. Bind and start to accept incoming connections.
        tcpChannelFuture = bootStrap.bind(port);    
        tcpChannelFuture.sync().addListener(FIRE_EXCEPTION_ON_FAILURE);  
        
		logger.info("server bounded. Protocol=" + protocol + ": " + port);
	}

	private void initializeSSLTCPTransport() throws InterruptedException {
		//1. load host and port
		String protocol = "SSL";
		Integer port = Configer.getInteger("ssl_port", null);
		
		if (port == null) {
			logger.info("skip server, Protocol=" + protocol);
			return;
		}
		
		//2. create bootstrap
		NettySSLChannelInitializer initializer = new NettySSLChannelInitializer();
        initializer.setMessageMetricsCollector(messageMetricsCollector);
        initializer.setBytesMetricsCollector(bytesMetricsCollector);
		
		ServerBootstrap bootStrap = new NettyServerBootstrap(this, initializer);

        //3. Bind and start to accept incoming connections.
		sslTCPChannelFuture = bootStrap.bind(port);    
		sslTCPChannelFuture.sync().addListener(FIRE_EXCEPTION_ON_FAILURE);  
        
		logger.info("server bounded. Protocol=" + protocol + ": " + port);
	}
	
	private void initializeWSSTransport() throws Exception {
		//1. load host and port 
		String protocol = "Secure Websocket";
		Integer port = Configer.getInteger("ws_port", null);
		
		if (port == null) {
			logger.info("skip server, Protocol=" + protocol);
			return;
		}
		
		//2. create bootstrap
        ChannelInitializer<SocketChannel> initializer = new NettyWSChannelInitializer();
        ServerBootstrap bootStrap = new NettyServerBootstrap(this, initializer);

        //3. Bind and start to accept incoming connections.
        wsChannelFuture = bootStrap.bind(port);    
        wsChannelFuture.sync().addListener(FIRE_EXCEPTION_ON_FAILURE);  
        
		logger.info("server bounded. Protocol=" + protocol + ": " + port);
	}
	
    private void closeOneChannel(ChannelFuture channelFuture) {
        if (channelFuture == null) {
        	return;
        }
        	
        Channel channel = channelFuture.channel();
        try {
        	channel.close();
	    } catch (Exception e) {
	    }    
        
        try {
        	channel.closeFuture().sync();
        } catch (Exception e) {
        }
	}
    
    private void closeOneEventLoopGroup(String name, EventLoopGroup loopGroup) {
        if (loopGroup == null) {
        	return;
        }
        
        //1. close gracefully
        Future<?> waiter = loopGroup.shutdownGracefully();

        logger.info("Waiting for" + name + " event loop groups to terminate...");
        try {
        	waiter.await(3, TimeUnit.SECONDS);
        } 
        catch (InterruptedException iex) {
        	logger.warn("An InterruptedException was caught while waiting for event loops to terminate...");
        }
        
        //2. close force
        if (!loopGroup.isTerminated()) {
        	logger.warn("Forcing shutdown " + name + "...");
        	loopGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS);
        }
	}

	public EventLoopGroup getBossGroup() {
		return bossGroup;
	}

	public EventLoopGroup getWorkerGroup() {
		return workerGroup;
	}

	public Class<? extends ServerSocketChannel> getChannelClass() {
		return channelClass;
	}

}