/*
 * Copyright (C) 2007-2019 Xagasoft, All rights reserved.
 *
 * This file is part of the libbu++ library and is released under the
 * terms of the license contained in the file LICENSE.
 */

#include "bu/server.h"
#include <errno.h>
#include <unistd.h>
#include "bu/serversocket.h"
#include "bu/client.h"
#include "bu/socket.h"
#include "bu/config.h"
#include "bu/mutexlocker.h"

#ifdef PROFILE_BU_SERVER
#define BU_PROFILE_START( x ) Bu::Profiler::getInstance().startEvent( x )
#define BU_PROFILE_END( x ) Bu::Profiler::getInstance().endEvent( x )
#else
#define BU_PROFILE_START( x ) (void)0
#define BU_PROFILE_END( x ) (void)0
#endif

#define RBS 1500

Bu::Server::Server( int iIoWorkers, int iClientWorkers ) :
    nTimeoutSec( 0 ),
    nTimeoutUSec( 0 ),
    bAutoTick( false )
{
    BU_PROFILE_START("server");
    FD_ZERO( &fdActive );

    if( iIoWorkers < 1 )
        iIoWorkers = 1;
    if( iClientWorkers < 1 )
        iClientWorkers = 1;

    for( int j = 0; j < iIoWorkers; j++ )
    {
        IoWorker *pWorker = new IoWorker( *this, qIoEvent, qClientEvent );
        lIoWorker.append( pWorker );
        pWorker->start();
    }

    for( int j = 0; j < iClientWorkers; j++ )
    {
        ClientWorker *pWorker = new ClientWorker( *this, qClientEvent );
        lClientWorker.append( pWorker );
        pWorker->start();
    }
}

Bu::Server::~Server()
{
    shutdown();
    BU_PROFILE_START("server");
}

void Bu::Server::addServerSocket( Bu::ServerSocket *pSocket )
{
    fd iFd;
    if( !pSocket->getFd( iFd ) )
    {
        throw Bu::ExceptionBase("Cannot get file descriptor from "
                "provided ServerSocket.");
    }
    FD_SET( iFd, &fdActive );
    hServers.insert( iFd, pSocket );
}

void Bu::Server::setTimeout( int nTimeoutSec, int nTimeoutUSec )
{
    this->nTimeoutSec = nTimeoutSec;
    this->nTimeoutUSec = nTimeoutUSec;
}

void Bu::Server::scan()
{
    BU_PROFILE_START("scan");
    struct timeval xTimeout = { nTimeoutSec, nTimeoutUSec };
    
    fd_set fdRead = fdActive;
    fd_set fdWrite /* = fdActive*/;
    fd_set fdException = fdActive;

    FD_ZERO( &fdWrite );
    mClients.lock();
    for( ClientHash::iterator i = hClients.begin(); i != hClients.end(); i++ )
    {
        if( (*i)->hasOutput() )
            FD_SET( i.getKey(), &fdWrite );
    }
    mClients.unlock();

    if( TEMP_FAILURE_RETRY( select( FD_SETSIZE,
            &fdRead, &fdWrite, &fdException, &xTimeout ) ) < 0 )
    {
        char buf[1024];
        strerror_r( errno, buf, 1024 );
        BU_PROFILE_END("scan");
        throw ExceptionBase(
            Bu::String("Error attempting to scan open connections: %1: %2").arg( errno ).arg( buf ).end().getStr()
            );
    }

    for( int j = 0; j < FD_SETSIZE; j++ )
    {
        if( FD_ISSET( j, &fdRead ) )
        {
            if( hServers.has( j ) )
            {
                Bu::ServerSocket *pSrv = hServers.get( j );
                addClient( pSrv, pSrv->accept() );
            }
            else
            {
                qIoEvent.enqueue( new Event( j, Event::Read ) );
            }
        }
        if( FD_ISSET( j, &fdWrite ) )
        {
            qIoEvent.enqueue( new Event( j, Event::Write ) );
        }
    }

    Bu::List<int> lDelete;
    // Now we just try to write all the pending data on all the sockets.
    // this could be done better eventually, if we care about the socket
    // wanting to accept writes (using a select).
    mClients.lock();
    for( ClientHash::iterator i = hClients.begin(); i != hClients.end(); i++ )
    {
        if( (*i)->wantsDisconnect() && !(*i)->hasOutput() )
        {
            lDelete.append( i.getKey() );
        }
    }
    mClients.unlock();

    for( Bu::List<int>::iterator i = lDelete.begin(); i != lDelete.end(); i++ )
    {
        closeClient( *i );
    }

    if( bAutoTick )
        tick();

    BU_PROFILE_END("scan");
}

void Bu::Server::addClient( const Bu::ServerSocket *pSrv, Bu::Socket *pSocket )
{
    BU_PROFILE_START("addClient");
    int iFdSrv;
    int iFdCli;
    if( !pSrv->getFd( iFdSrv ) || !pSrv->getFd( iFdCli ) )
    {
        throw Bu::ExceptionBase("No file descriptor?");
    }
    FD_SET( iFdCli, &fdActive );

    Client *pClient = new Client(
        new SrvClientLinkFactory()
        );
    {
        Bu::MutexLocker l( mClients );
        hClients.insert( iFdCli, pClient );
        hSockets.insert( iFdCli, pSocket );
    }

    onNewConnection( pSrv, pClient, pSocket );
    BU_PROFILE_END("addClient");
}

Bu::Client *Bu::Server::getClient( fd iId )
{
    Bu::MutexLocker l( mClients );
    return hClients.get( iId );
}

bool Bu::Server::getClientAndSocket( fd iId, Bu::Client *&pClient,
        Bu::Socket *&pSocket )
{
    Bu::MutexLocker l( mClients );
    if( !hClients.has( iId ) || !hSockets.has( iId ) )
        return false;
    pClient = hClients.get( iId );
    pSocket = hSockets.get( iId );
    return true;
}

Bu::Server::SrvClientLink::SrvClientLink( Bu::Client *pClient ) :
    pClient( pClient )
{
}

Bu::Server::SrvClientLink::~SrvClientLink()
{
}

void Bu::Server::SrvClientLink::sendMessage( const Bu::String &sMsg )
{
    pClient->onMessage( sMsg );
}

Bu::Server::SrvClientLinkFactory::SrvClientLinkFactory()
{
}

Bu::Server::SrvClientLinkFactory::~SrvClientLinkFactory()
{
}

Bu::ClientLink *Bu::Server::SrvClientLinkFactory::createLink(
        Bu::Client *pClient )
{
    return new SrvClientLink( pClient );
}

void Bu::Server::setAutoTick( bool bEnable )
{
    bAutoTick = bEnable;
}

void Bu::Server::tick()
{
    mClients.lock();
    for( ClientHash::iterator i = hClients.begin(); i != hClients.end(); i++ )
    {
        (*i)->tick();
    }
    mClients.unlock();
}

void Bu::Server::shutdown()
{
    {
        qIoEvent.stop();
        qClientEvent.stop();
        Bu::Server::Event *pEv;
        while( (pEv = qIoEvent.drain()) != NULL )
        {
            delete pEv;
        }
        while( (pEv = qClientEvent.drain()) != NULL )
        {
            delete pEv;
        }

        Bu::MutexLocker l( mWorkers );
        for( IoWorkerList::iterator i = lIoWorker.begin(); i; i++ )
        {
            (*i)->join();
            delete *i;
        }
        lIoWorker.clear();
        for( ClientWorkerList::iterator i = lClientWorker.begin();
                i; i++ )
        {
            (*i)->join();
            delete *i;
        }
        lClientWorker.clear();
    }

    for( SrvHash::iterator i = hServers.begin(); i != hServers.end(); i++ )
    {
        delete *i;
    }

    hServers.clear();

    ClientHash::KeyList lClients = hClients.getKeys();
    for( ClientHash::KeyList::iterator i = lClients.begin(); i; i++ )
    {
        closeClient( *i );
    }
    hClients.clear();
}

void Bu::Server::closeClient( fd iSocket )
{
    BU_PROFILE_START("closeClient");
    Bu::Client *pClient = hClients.get( iSocket );
    Bu::Socket *pSocket = hSockets.get( iSocket );
    onClosedConnection( pClient );
    pClient->close();
    hClients.erase( iSocket );
    pSocket->close();
    hSockets.erase( iSocket );
    FD_CLR( iSocket, &fdActive );
    delete pClient;
    delete pSocket;
    BU_PROFILE_END("closeClient");
}

////////
// Event
////

Bu::Server::Event::Event( fd iId, Operation eOp ) :
    iId( iId ),
    eOp( eOp )
{
}

Bu::Server::Event::~Event()
{
}

Bu::Server::fd Bu::Server::Event::getId() const
{
    return iId;
}

Bu::Server::Event::Operation Bu::Server::Event::getOperation() const
{
    return eOp;
}

/////////
// IoWorker
////

Bu::Server::IoWorker::IoWorker( Bu::Server &rSrv,
        Bu::Server::EventQueue &qIoEvent,
        Bu::Server::EventQueue &qClientEvent ) :
    rSrv( rSrv ),
    qIoEvent( qIoEvent ),
    qClientEvent( qClientEvent )
{
}

Bu::Server::IoWorker::~IoWorker()
{
}

void Bu::Server::IoWorker::run()
{
    while( qIoEvent.isRunning() )
    {
        Event *pEv = qIoEvent.dequeue();
        if( pEv == NULL )
            continue;

        Client *pClient;
        Socket *pSocket;
        if( !rSrv.getClientAndSocket( pEv->getId(), pClient, pSocket ) )
        {
            delete pEv;
            continue;
        }

        switch( pEv->getOperation() )
        {
            case Event::Read:
                handleRead( pClient, pSocket );
                break;

            case Event::Write:
                handleWrite( pClient, pSocket );
                break;

            case Event::Process:
                break;
        }
        
        delete pEv;
    }
}

void Bu::Server::IoWorker::handleRead( Client *pClient, Socket *pSocket )
{
    char buf[RBS];
    Bu::size iRead;
    Bu::size iTotal=0;

    BU_PROFILE_START("client.read");
    for(;;)
    {
        try
        {
            iRead = pSocket->read( buf, RBS );

            if( iRead == 0 )
            {
                break;
            }
            else
            {
                iTotal += iRead;
                pClient->cbBuffer.server().write( buf, iRead );
                if( !pSocket->canRead() )
                    break;
            }
        }
        catch( Bu::ExceptionBase &e )
        {
            close( pSocket );
            break;
        }
    }
    BU_PROFILE_END("client.read");

    if( iTotal == 0 )
    {
        close( pSocket );
    }
    else
    {
        Bu::Server::fd iFd;
        pSocket->getFd( iFd );
        qClientEvent.enqueue( new Event( iFd, Event::Process ) );
    }
}

void Bu::Server::IoWorker::handleWrite( Client *pClient, Socket *pSocket )
{
    char buf[RBS];
    if( pClient->hasOutput() > 0 )
    {
        int iAmnt = RBS;
        iAmnt = pClient->cbBuffer.server().peek( buf, iAmnt );
        int iReal = pSocket->write( buf, iAmnt );
        pClient->cbBuffer.server().seek( iReal );
    }
}

void Bu::Server::IoWorker::close( Bu::Socket *pSocket )
{
    Bu::Server::fd iFd;
    pSocket->getFd( iFd );
    rSrv.closeClient( iFd );
}           

/////////
// ClientWorker
////

Bu::Server::ClientWorker::ClientWorker( Bu::Server &rSrv,
        Bu::Server::EventQueue &qEvent ) :
    rSrv( rSrv ),
    qEvent( qEvent )
{
}

Bu::Server::ClientWorker::~ClientWorker()
{
}

void Bu::Server::ClientWorker::run()
{
    while( qEvent.isRunning() )
    {
        Event *pEv = qEvent.dequeue();
        if( pEv == NULL )
            continue;

        Client *pClient = rSrv.getClient( pEv->getId() );
        if( pClient == NULL )
        {
            delete pEv;
            continue;
        }
        
        pClient->processInput();
        delete pEv;
    }
}