/*
 * Collie - An asynchronous event-driven network framework using Dlang development
 *
 * Copyright (C) 2015-2017  Shanghai Putao Technology Co., Ltd 
 *
 * Developer: putao's Dlang team
 *
 * Licensed under the Apache-2.0 License.
 *
 */
module collie.bootstrap.server;
import kiss.log;
import collie.net;
import collie.channel;
import collie.bootstrap.serversslconfig;
public import collie.bootstrap.exception;


import std.exception;

final class ServerBootstrap(PipeLine)
{
    this()
    {
        _loop = new EventLoop();
    }

    this(EventLoop loop)
    {
        _loop = loop;
    }

    auto pipeline(shared AcceptPipelineFactory factory)
    {
        _acceptorPipelineFactory = factory;
        return this;
    }

    auto setSSLConfig(ServerSSLConfig config)
    {
	       _sslConfig = config;
	       return this;
    }

    auto childPipeline(shared PipelineFactory!PipeLine factory)
    {
        _childPipelineFactory = factory;
        return this;
    }

    auto group(EventLoopGroup group)
    {
        _group = group;
        return this;
    }

    auto setReusePort(bool ruse)
    {
        _rusePort = ruse;
        return this;
    }

    /**
            The Value will be 0 or 5s ~ 1800s.
            0 is disable, 
            if(value < 5) value = 5;
            if(value > 3000) value = 1800;
        */
    auto heartbeatTimeOut(uint second)
    {
        _timeOut = second;
        _timeOut = _timeOut < 5 ? 5 : _timeOut;
        _timeOut = _timeOut > 1800 ? 1800 : _timeOut;

        return this;
    }

    void bind(Address addr)
    {
        _address = addr;
    }

    void bind(ushort port)
    {
        _address = new InternetAddress(port);
    }

    void bind(string ip, ushort port)
    {
        _address = new InternetAddress(ip, port);
    }

    void stopListening()
    {
		if (!_listening)
            return;
		scope(exit)_listening = false;
        foreach (ref accept; _serverlist)
        {
            accept.stop();
        }
        _mainAccept.stop();

    }

	void stop()
	{
		if(!_isLoopWait) return;
		scope(exit) _isLoopWait = false;
		_group.stop();
		_loop.stop();
	}

    void join()
    {
		if (!_isLoopWait)
            return;
        if (_group)
            _group.wait();
    }

    void waitForStop()
    {
		if(_isLoopWait)
			throw new ServerIsRuningException("server is runing!");
		if(!_listening)
			startListening();
		_isLoopWait = true;
		if(_group)
			_group.start();
		_loop.join();
    }

	void startListening()
	{
		if (_listening)
			throw new ServerIsListeningException("server is listening!");
		if (_address is null || _childPipelineFactory is null)
			throw new ServerStartException("the address or childPipelineFactory is null!");

		_listening = true;
		uint wheel, time;
		bool beat = getTimeWheelConfig(wheel, time);
		_mainAccept = creatorAcceptor(_loop);
		_mainAccept.initialize();
		if (beat)
			_mainAccept.startTimingWhile(wheel, time);
		if (_group)
		{
			foreach (loop; _group)
			{
				auto acceptor = creatorAcceptor(loop);
				acceptor.initialize();
				_serverlist ~= acceptor;
				if (beat)
					acceptor.startTimingWhile(wheel, time);
			}
		}
		logDebug("server _listening!");
	}

	EventLoopGroup group(){return _group;}

	@property EventLoop eventLoop(){return _loop;}

	@property Address address(){return _address;}

protected:
    auto creatorAcceptor(EventLoop loop)
    {
        auto acceptor = new TcpListener(loop, _address.addressFamily);
		if(_rusePort)
        	acceptor.reusePort = _rusePort;
        acceptor.bind(_address);
        acceptor.listen(1024);
		{
			Linger optLinger;
			optLinger.on = 1;
			optLinger.time = 0;
			acceptor.setOption(SocketOptionLevel.SOCKET, SocketOption.LINGER, optLinger);
		}
        AcceptPipeline pipe;
        if (_acceptorPipelineFactory)
            pipe = _acceptorPipelineFactory.newPipeline(acceptor);
        else
            pipe = AcceptPipeline.create();

        SSL_CTX* ctx = null;
		version(USE_SSL)
        {
            if (_sslConfig)
            {
                ctx = _sslConfig.generateSSLCtx();
                if (!ctx)
					throw new SSLException("Can not gengrate SSL_CTX");
            }
        }

        return new ServerAcceptor!(PipeLine)(acceptor, pipe, _childPipelineFactory,
            ctx);
    }

    bool getTimeWheelConfig(out uint whileSize, out uint time)
    {
        if (_timeOut == 0)
            return false;
        if (_timeOut <= 40)
        {
            whileSize = 50;
            time = _timeOut * 1000 / 50;
        }
        else if (_timeOut <= 120)
        {
            whileSize = 60;
            time = _timeOut * 1000 / 60;
        }
        else if (_timeOut <= 600)
        {
            whileSize = 100;
            time = _timeOut * 1000 / 100;
        }
        else if (_timeOut < 1000)
        {
            whileSize = 150;
            time = _timeOut * 1000 / 150;
        }
        else
        {
            whileSize = 180;
            time = _timeOut * 1000 / 180;
        }
        return true;
    }

private:
    shared AcceptPipelineFactory _acceptorPipelineFactory;
    shared PipelineFactory!PipeLine _childPipelineFactory;

    ServerAcceptor!(PipeLine) _mainAccept;
    EventLoop _loop;

    ServerAcceptor!(PipeLine)[] _serverlist;
    EventLoopGroup _group;

	bool _listening = false;
    bool _rusePort = true;
	bool _isLoopWait = false;
    uint _timeOut = 0;
    Address _address;

    ServerSSLConfig _sslConfig = null;
}

private:

import std.functional;
import kiss.timingwheel;
import collie.utils.memory;
import collie.net;

final @trusted class ServerAcceptor(PipeLine) : InboundHandler!(Socket)
{
    this(TcpListener acceptor, AcceptPipeline pipe,
        shared PipelineFactory!PipeLine clientPipeFactory, SSL_CTX* ctx = null)
    {
        _acceptor = acceptor;
        _pipeFactory = clientPipeFactory;
        pipe.addBack(this);
        pipe.finalize();
        _pipe = pipe;
        _pipe.transport(_acceptor);
        _acceptor.setReadHandle(&acceptCallBack);
        _sslctx = ctx;
		_list = new ServerConnection!PipeLine();
		version(USE_SSL)
			_sharkList = new SSLHandShark();
    }

    pragma(inline, true) void initialize()
    {
        _pipe.transportActive();
    }

    pragma(inline, true) void stop()
    {
        _pipe.transportInactive();
    }

    override void read(Context ctx, Socket msg)
    {
		version(USE_SSL)
        {
            if (_sslctx)
            {
                    auto ssl = SSL_new(_sslctx);
				static if (IOMode == IO_MODE.iocp){
					BIO * readBIO = BIO_new(BIO_s_mem());
					BIO * writeBIO = BIO_new(BIO_s_mem());
					SSL_set_bio(ssl, readBIO, writeBIO);
					SSL_set_accept_state(ssl);
					auto asynssl = new SSLSocket(_acceptor.eventLoop, msg, ssl,readBIO,writeBIO);
				} else {
                    if (SSL_set_fd(ssl, msg.handle()) < 0)
                    {
                        error("SSL_set_fd error: fd = ", msg.handle());
                        SSL_shutdown(ssl);
                        SSL_free(ssl);
                        return;
                    }
                    SSL_set_accept_state(ssl);
                    auto asynssl = new SSLSocket(_acceptor.eventLoop, msg, ssl);
				}
                    auto shark = new SSLHandShark(asynssl, &doHandShark);

					shark.next = _sharkList.next;
					if(shark.next) shark.next.prev = shark;
					shark.prev = _sharkList;
					_sharkList.next = shark;

                    asynssl.start();
            }
            else
            {
                auto asyntcp = new TcpStream(_acceptor.eventLoop, msg);
                startSocket(asyntcp);
            }
        } else
        {
            auto asyntcp = new TcpStream(_acceptor.eventLoop, msg);
            startSocket(asyntcp);
        }
    }

    override void transportActive(Context ctx)
    {
        logDebug("acept transportActive");
        if (!_acceptor.watch())
        {
            logError("acceptor start error!");
        }
    }

    override void transportInactive(Context ctx)
    {
        _acceptor.close();
		auto con = _list.next;
		_list.next = null;
		while(con) {
			auto tcon = con;
			con = con.next;
			tcon.close();
		}
        _acceptor.eventLoop.stop();
    }

protected:
    pragma(inline) void remove(ServerConnection!PipeLine conn)
    {
		conn.prev.next = conn.next;
		if(conn.next)
			conn.next.prev = conn.prev;
        gcFree(conn);
    }

    void acceptCallBack(EventLoop loop, Socket socket)  nothrow 
    {
        catchAndLogException(_pipe.read(socket));
    }

    pragma(inline, true) @property acceptor()
    {
        return _acceptor;
    }

    void startTimingWhile(uint whileSize, uint time)
    {
        if (_timer)
            return;
        _timer = new Timer(_acceptor.eventLoop);
        _timer.setTimerHandle(&doWheel);
        _wheel = new TimingWheel(whileSize);
        _timer.start(time);
    }

    void doWheel() nothrow
    {
        if (_wheel)
            _wheel.prevWheel();
    }

	version(USE_SSL)
    {
        void doHandShark(SSLHandShark shark, SSLSocket sock)
        {
			shark.prev.next = shark.next;
			if(shark.next) shark.next.prev = shark.prev;
            scope (exit)
                delete shark;
            if (sock)
            {
                sock.setHandshakeCallBack(null);
                startSocket(sock);
            }
        }
    }

    void startSocket(TcpStream sock)
    {
        auto pipe = _pipeFactory.newPipeline(sock);
        if (!pipe)
        {
            gcFree(sock);
            return;
        }
        pipe.finalize();
        auto con = new ServerConnection!PipeLine(pipe);
        con.serverAceptor = this;

		con.next = _list.next;
		if(con.next)
			con.next.prev = con;
		con.prev = _list;
		_list.next = con;

        con.initialize();
        if (_wheel)
            _wheel.addNewTimer(con);
    }

private:
   // int[ServerConnection!PipeLine] _list;
	ServerConnection!PipeLine _list;

	version(USE_SSL)
    {
		SSLHandShark _sharkList;
    }

    TcpListener _acceptor;
    Timer _timer;
    TimingWheel _wheel;
    AcceptPipeline _pipe;
    shared PipelineFactory!PipeLine _pipeFactory;

    SSL_CTX* _sslctx = null;
}

@trusted final class ServerConnection(PipeLine) : WheelTimer, PipelineManager
{
    this(PipeLine pipe)
    {
        _pipe = pipe;
        _pipe.pipelineManager = this;
    }

    ~this()
    {
    }

    pragma(inline, true) void initialize()
    {
        _pipe.transportActive();
    }

    pragma(inline, true) void close()
    {
        _pipe.transportInactive();
    }

    pragma(inline, true) @property serverAceptor()
    {
        return _manger;
    }

    pragma(inline, true) @property serverAceptor(ServerAcceptor!PipeLine manger)
    {
        _manger = manger;
    }

    override void deletePipeline(PipelineBase pipeline)
    {
        pipeline.pipelineManager = null;
        _pipe = null;
        stop();
        _manger.remove(this);
    }

    override void refreshTimeout()
    {
        rest();
    }

    override void onTimeOut() nothrow
    {
		collectException(_pipe.timeOut());
    }
private:
	this(){}
	ServerConnection!PipeLine prev;
	ServerConnection!PipeLine next;
private:
    ServerAcceptor!PipeLine _manger;
    PipeLine _pipe;
}

version(USE_SSL)
{
    final class SSLHandShark
    {
        alias SSLHandSharkCallBack = void delegate(SSLHandShark shark, SSLSocket sock);
        this(SSLSocket sock, SSLHandSharkCallBack cback)
        {
            _socket = sock;
            _cback = cback;
            _socket.setCloseCallBack(&onClose);
            _socket.setReadCallBack(&readCallBack);
            _socket.setHandshakeCallBack(&handSharkCallBack);
        }

        ~this()
        {
        }

    protected:
        void handSharkCallBack()
        {
            logDebug("the ssl handshark over");
            _cback(this, _socket);
            _socket = null;
        }

        void readCallBack(ubyte[] buffer)
        {
        }

        void onClose()
        {
            logDebug("the ssl handshark fail");
            _socket.setCloseCallBack(null);
            _socket.setReadCallBack(null);
            _socket.setHandshakeCallBack(null);
            _socket = null;
            _cback(this, _socket);
        }

	private:
		this(){}
		SSLHandShark prev;
		SSLHandShark next;
    private:
        SSLSocket _socket;
        SSLHandSharkCallBack _cback;
    }
}