/*
 * Copyright 2007 Stephen Liu
 * For license terms, see the file COPYING along with this library.
 */

#include <fcntl.h>
#include <stdio.h>
#include <string.h>
#include <assert.h>
#include <errno.h>
#include <stdlib.h>

#include "spporting.hpp"

#include "speventcb.hpp"
#include "spexecutor.hpp"
#include "spsession.hpp"
#include "spresponse.hpp"
#include "sphandler.hpp"
#include "spbuffer.hpp"
#include "spmsgdecoder.hpp"
#include "sputils.hpp"
#include "sprequest.hpp"
#include "spmsgblock.hpp"
#include "spiochannel.hpp"
#include "spioutils.hpp"

#include "event_msgqueue.h"
#include "event.h"

SP_EventArg :: SP_EventArg( int timeout )
{
	mEventBase = (struct event_base*)event_init();

	mResponseQueue = msgqueue_new( mEventBase, 0,
			SP_EventCallback::onResponse, this );

	mInputResultQueue = new SP_BlockingQueue();

	mOutputResultQueue = new SP_BlockingQueue();

	mSessionManager = new SP_SessionManager();

	mTimeout = timeout;
}

SP_EventArg :: ~SP_EventArg()
{
	delete mInputResultQueue;
	delete mOutputResultQueue;

	delete mSessionManager;

	//msgqueue_destroy( (struct event_msgqueue*)mResponseQueue );
	//event_base_free( mEventBase );
}

struct event_base * SP_EventArg :: getEventBase() const
{
	return mEventBase;
}

void * SP_EventArg :: getResponseQueue() const
{
	return mResponseQueue;
}

SP_BlockingQueue * SP_EventArg :: getInputResultQueue() const
{
	return mInputResultQueue;
}

SP_BlockingQueue * SP_EventArg :: getOutputResultQueue() const
{
	return mOutputResultQueue;
}

SP_SessionManager * SP_EventArg :: getSessionManager() const
{
	return mSessionManager;
}

void SP_EventArg :: setTimeout( int timeout )
{
	mTimeout = timeout;
}

int SP_EventArg :: getTimeout() const
{
	return mTimeout;
}

//-------------------------------------------------------------------

void SP_EventCallback :: onAccept( int fd, short events, void * arg )
{
	int clientFD;
	struct sockaddr_in addr;
	socklen_t addrLen = sizeof( addr );

	SP_AcceptArg_t * acceptArg = (SP_AcceptArg_t*)arg;
	SP_EventArg * eventArg = acceptArg->mEventArg;

	clientFD = accept( fd, (struct sockaddr *)&addr, &addrLen );
	if( -1 == clientFD ) {
		sp_syslog( LOG_WARNING, "accept failed" );
		return;
	}

	if( SP_IOUtils::setNonblock( clientFD ) < 0 ) {
		sp_syslog( LOG_WARNING, "failed to set client socket non-blocking" );
	}

	SP_Sid_t sid;
	sid.mKey = eventArg->getSessionManager()->allocKey( &sid.mSeq );
	assert( sid.mKey > 0 );

	SP_Session * session = new SP_Session( sid );

	char strip[ 32 ] = { 0 };
	SP_IOUtils::inetNtoa( &( addr.sin_addr ), strip, sizeof( strip ) );
	session->getRequest()->setClientIP( strip );
	session->getRequest()->setClientPort( ntohs( addr.sin_port ) );

	if( 0 == getsockname( clientFD, (struct sockaddr*)&addr, &addrLen ) ) {
		SP_IOUtils::inetNtoa( &( addr.sin_addr ), strip, sizeof( strip ) );
		session->getRequest()->setServerIP( strip );
	}

	if( NULL != session ) {
		eventArg->getSessionManager()->put( sid.mKey, sid.mSeq, session );

		session->setHandler( acceptArg->mHandlerFactory->create() );
		session->setIOChannel( acceptArg->mIOChannelFactory->create() );
		session->setArg( eventArg );

		event_set( session->getReadEvent(), clientFD, EV_READ, onRead, session );
		event_set( session->getWriteEvent(), clientFD, EV_WRITE, onWrite, session );

		if( eventArg->getSessionManager()->getCount() > acceptArg->mMaxConnections
				|| eventArg->getInputResultQueue()->getLength() >= acceptArg->mReqQueueSize ) {
			sp_syslog( LOG_WARNING, "System busy, session.count %d [%d], queue.length %d [%d]",
				eventArg->getSessionManager()->getCount(), acceptArg->mMaxConnections,
				eventArg->getInputResultQueue()->getLength(), acceptArg->mReqQueueSize );

			SP_Message * msg = new SP_Message();
			msg->getMsg()->append( acceptArg->mRefusedMsg );
			msg->getMsg()->append( "\r\n" );
			session->getOutList()->append( msg );
			session->setStatus( SP_Session::eExit );

			addEvent( session, EV_WRITE, clientFD );
		} else {
			SP_EventHelper::doStart( session );
		}
	} else {
		eventArg->getSessionManager()->remove( sid.mKey, sid.mSeq );
		sp_close( clientFD );
		sp_syslog( LOG_WARNING, "Out of memory, cannot allocate session object!" );
	}
}

void SP_EventCallback :: onRead( int fd, short events, void * arg )
{
	SP_Session * session = (SP_Session*)arg;

	session->setReading( 0 );

	SP_Sid_t sid = session->getSid();

	if( EV_READ & events ) {
		int len = session->getIOChannel()->receive( session );

		if( len > 0 ) {
			session->addRead( len );
			if( 0 == session->getRunning() ) {
				SP_MsgDecoder * decoder = session->getRequest()->getMsgDecoder();
				if( SP_MsgDecoder::eOK == decoder->decode( session->getInBuffer() ) ) {
					SP_EventHelper::doWork( session );
				}
			}
			addEvent( session, EV_READ, -1 );
		} else {
			int saved = errno;

			if( 0 != errno ) {
				sp_syslog( LOG_WARNING, "session(%d.%d) read error, errno %d, status %d",
						sid.mKey, sid.mSeq, errno, session->getStatus() );
			}

			if( EAGAIN != saved ) {
				if( 0 == session->getRunning() ) {
					SP_EventHelper::doError( session );
				} else {
					sp_syslog( LOG_NOTICE, "session(%d.%d) busy, process session error later",
							sid.mKey, sid.mSeq );
					// If this session is running, then onResponse will add write event for this session.
					// It will be processed as write fail at the last. So no need to re-add event here.
				}
			} else {
				addEvent( session, EV_READ, -1 );
			}
		}
	} else {
		if( 0 == session->getRunning() ) {
			SP_EventHelper::doTimeout( session );
		} else {
			sp_syslog( LOG_NOTICE, "session(%d.%d) busy, process session timeout later",
					sid.mKey, sid.mSeq );
			// If this session is running, then onResponse will add write event for this session.
			// It will be processed as write fail at the last. So no need to re-add event here.
		}
	}
}

void SP_EventCallback :: onWrite( int fd, short events, void * arg )
{
	SP_Session * session = (SP_Session*)arg;

	session->setWriting( 0 );

	SP_Sid_t sid = session->getSid();

	if( EV_WRITE & events ) {
		int ret = 0;

		if( session->getOutList()->getCount() > 0 ) {
			int len = session->getIOChannel()->transmit( session );
			if( len > 0 ) {
				session->addWrite( len );
				if( session->getOutList()->getCount() > 0 ) {
					// left for next write event
					addEvent( session, EV_WRITE, -1 );
				}
			} else {
				if( EAGAIN != errno ) {
					ret = -1;
					if( 0 == session->getRunning() ) {
						sp_syslog( LOG_NOTICE, "session(%d.%d) write error, errno %d, status %d, count %d",
								sid.mKey, sid.mSeq, errno, session->getStatus(), session->getOutList()->getCount() );
						SP_EventHelper::doError( session );
					} else {
						sp_syslog( LOG_NOTICE, "session(%d.%d) busy, process session error later, errno [%d]",
								sid.mKey, sid.mSeq, errno );
						// If this session is running, then onResponse will add write event for this session.
						// It will be processed as write fail at the last. So no need to re-add event here.
					}
				} else {
					addEvent( session, EV_WRITE, -1 );
				}
			}
		}

		if( 0 == ret && session->getOutList()->getCount() <= 0 ) {
			if( SP_Session::eExit == session->getStatus() ) {
				ret = -1;
				if( 0 == session->getRunning() ) {
					sp_syslog( LOG_DEBUG, "session(%d.%d) normal exit", sid.mKey, sid.mSeq );
					SP_EventHelper::doClose( session );
				} else {
					sp_syslog( LOG_NOTICE, "session(%d.%d) busy, terminate session later",
							sid.mKey, sid.mSeq );
					// If this session is running, then onResponse will add write event for this session.
					// It will be processed as write fail at the last. So no need to re-add event here.
				}
			}
		}

		if( 0 == ret ) {
			if( 0 == session->getRunning() ) {
				SP_MsgDecoder * decoder = session->getRequest()->getMsgDecoder();
				if( SP_MsgDecoder::eOK == decoder->decode( session->getInBuffer() ) ) {
					SP_EventHelper::doWork( session );
				}
			} else {
				// If this session is running, then onResponse will add write event for this session.
				// So no need to add write event here.
			}
		}
	} else {
		if( 0 == session->getRunning() ) {
			SP_EventHelper::doTimeout( session );
		} else {
			sp_syslog( LOG_NOTICE, "session(%d.%d) busy, process session timeout later",
					sid.mKey, sid.mSeq );
			// If this session is running, then onResponse will add write event for this session.
			// It will be processed as write fail at the last. So no need to re-add event here.
		}
	}
}

void SP_EventCallback :: onResponse( void * queueData, void * arg )
{
	SP_Response * response = (SP_Response*)queueData;
	SP_EventArg * eventArg = (SP_EventArg*)arg;
	SP_SessionManager * manager = eventArg->getSessionManager();

	SP_Sid_t fromSid = response->getFromSid();
	uint16_t seq = 0;

	if( ! SP_EventHelper::isSystemSid( &fromSid ) ) {
		SP_Session * session = manager->get( fromSid.mKey, &seq );
		if( seq == fromSid.mSeq && NULL != session ) {
			if( SP_Session::eWouldExit == session->getStatus() ) {
				session->setStatus( SP_Session::eExit );
			}

			if( SP_Session::eNormal != session->getStatus() ) {
				event_del( session->getReadEvent() );
			}

			// always add a write event for sender, 
			// so the pending input can be processed in onWrite
			addEvent( session, EV_WRITE, -1 );
			addEvent( session, EV_READ, -1 );
		} else {
			sp_syslog( LOG_WARNING, "session(%d.%d) invalid, unknown FROM",
					fromSid.mKey, fromSid.mSeq );
		}
	}

	for( SP_Message * msg = response->takeMessage();
			NULL != msg; msg = response->takeMessage() ) {

		SP_SidList * sidList = msg->getToList();

		if( msg->getTotalSize() > 0 ) {
			for( int i = sidList->getCount() - 1; i >= 0; i-- ) {
				SP_Sid_t sid = sidList->get( i );
				SP_Session * session = manager->get( sid.mKey, &seq );
				if( seq == sid.mSeq && NULL != session ) {
					if( 0 != memcmp( &fromSid, &sid, sizeof( sid ) )
							&& SP_Session::eExit == session->getStatus() ) {
						sidList->take( i );
						msg->getFailure()->add( sid );
						sp_syslog( LOG_WARNING, "session(%d.%d) would exit, invalid TO", sid.mKey, sid.mSeq );
					} else {
						session->getOutList()->append( msg );
						addEvent( session, EV_WRITE, -1 );
					}
				} else {
					sidList->take( i );
					msg->getFailure()->add( sid );
					sp_syslog( LOG_WARNING, "session(%d.%d) invalid, unknown TO", sid.mKey, sid.mSeq );
				}
			}
		} else {
			for( ; sidList->getCount() > 0; ) {
				msg->getFailure()->add( sidList->take( SP_ArrayList::LAST_INDEX ) );
			}
		}

		if( msg->getToList()->getCount() <= 0 ) {
			SP_EventHelper::doCompletion( eventArg, msg );
		}
	}

	for( int i = 0; i < response->getToCloseList()->getCount(); i++ ) {
		SP_Sid_t sid = response->getToCloseList()->get( i );
		SP_Session * session = manager->get( sid.mKey, &seq );
		if( seq == sid.mSeq && NULL != session ) {
			session->setStatus( SP_Session::eExit );
			addEvent( session, EV_WRITE, -1 );
		} else {
			sp_syslog( LOG_WARNING, "session(%d.%d) invalid, unknown CLOSE", sid.mKey, sid.mSeq );
		}
	}

	delete response;
}

void SP_EventCallback :: addEvent( SP_Session * session, short events, int fd )
{
	SP_EventArg * eventArg = (SP_EventArg*)session->getArg();

	if( ( events & EV_WRITE ) && 0 == session->getWriting() ) {
		session->setWriting( 1 );

		if( fd < 0 ) fd = EVENT_FD( session->getWriteEvent() );

		event_set( session->getWriteEvent(), fd, events, onWrite, session );
		event_base_set( eventArg->getEventBase(), session->getWriteEvent() );

		struct timeval timeout;
		memset( &timeout, 0, sizeof( timeout ) );
		timeout.tv_sec = eventArg->getTimeout();
		event_add( session->getWriteEvent(), &timeout );
	}

	if( events & EV_READ && 0 == session->getReading() ) {
		session->setReading( 1 );

		if( fd < 0 ) fd = EVENT_FD( session->getWriteEvent() );

		event_set( session->getReadEvent(), fd, events, onRead, session );
		event_base_set( eventArg->getEventBase(), session->getReadEvent() );

		struct timeval timeout;
		memset( &timeout, 0, sizeof( timeout ) );
		timeout.tv_sec = eventArg->getTimeout();
		event_add( session->getReadEvent(), &timeout );
	}
}

//-------------------------------------------------------------------

int SP_EventHelper :: isSystemSid( SP_Sid_t * sid )
{
	return ( sid->mKey == SP_Sid_t::eTimerKey && sid->mSeq == SP_Sid_t::eTimerSeq )
			|| ( sid->mKey == SP_Sid_t::ePushKey && sid->mSeq == SP_Sid_t::ePushSeq );
}

void SP_EventHelper :: doWork( SP_Session * session )
{
	if( SP_Session::eNormal == session->getStatus() ) {
		session->setRunning( 1 );
		SP_EventArg * eventArg = (SP_EventArg*)session->getArg();
		eventArg->getInputResultQueue()->push( new SP_SimpleTask( worker, session, 1 ) );
	} else {
		SP_Sid_t sid = session->getSid();

		char buffer[ 16 ] = { 0 };
		session->getInBuffer()->take( buffer, sizeof( buffer ) );
		sp_syslog( LOG_WARNING, "session(%d.%d) status is %d, ignore [%s...] (%dB)",
			sid.mKey, sid.mSeq, session->getStatus(), buffer, session->getInBuffer()->getSize() );
		session->getInBuffer()->reset();
	}
}

void SP_EventHelper :: worker( void * arg )
{
	SP_Session * session = (SP_Session*)arg;
	SP_Handler * handler = session->getHandler();
	SP_EventArg * eventArg = (SP_EventArg *)session->getArg();

	SP_Response * response = new SP_Response( session->getSid() );
	if( 0 != handler->handle( session->getRequest(), response ) ) {
		session->setStatus( SP_Session::eWouldExit );
	}

	session->setRunning( 0 );

	msgqueue_push( (struct event_msgqueue*)eventArg->getResponseQueue(), response );
}

void SP_EventHelper :: doError( SP_Session * session )
{
	SP_EventArg * eventArg = (SP_EventArg *)session->getArg();

	event_del( session->getWriteEvent() );
	event_del( session->getReadEvent() );

	SP_Sid_t sid = session->getSid();

	SP_ArrayList * outList = session->getOutList();
	for( ; outList->getCount() > 0; ) {
		SP_Message * msg = ( SP_Message * ) outList->takeItem( SP_ArrayList::LAST_INDEX );

		int index = msg->getToList()->find( sid );
		if( index >= 0 ) msg->getToList()->take( index );
		msg->getFailure()->add( sid );

		if( msg->getToList()->getCount() <= 0 ) {
			doCompletion( eventArg, msg );
		}
	}

	// remove session from SessionManager, onResponse will ignore this session
	eventArg->getSessionManager()->remove( sid.mKey, sid.mSeq );

	eventArg->getInputResultQueue()->push( new SP_SimpleTask( error, session, 1 ) );
}

void SP_EventHelper :: error( void * arg )
{
	SP_Session * session = ( SP_Session * )arg;
	SP_EventArg * eventArg = (SP_EventArg*)session->getArg();

	SP_Sid_t sid = session->getSid();

	SP_Response * response = new SP_Response( sid );
	session->getHandler()->error( response );

	msgqueue_push( (struct event_msgqueue*)eventArg->getResponseQueue(), response );

	sp_syslog( LOG_WARNING, "session(%d.%d) error, r %d(%d), w %d(%d), i %d, o %d, s %d(%d)",
			sid.mKey, sid.mSeq, session->getTotalRead(), session->getReading(),
			session->getTotalWrite(), session->getWriting(),
			session->getInBuffer()->getSize(), session->getOutList()->getCount(),
			eventArg->getSessionManager()->getCount(), eventArg->getSessionManager()->getFreeCount() );

	// onResponse will ignore this session, so it's safe to destroy session here
	session->getHandler()->close();
	sp_close( EVENT_FD( session->getWriteEvent() ) );
	delete session;
}

void SP_EventHelper :: doTimeout( SP_Session * session )
{
	SP_EventArg * eventArg = (SP_EventArg*)session->getArg();

	event_del( session->getWriteEvent() );
	event_del( session->getReadEvent() );

	SP_Sid_t sid = session->getSid();

	SP_ArrayList * outList = session->getOutList();
	for( ; outList->getCount() > 0; ) {
		SP_Message * msg = ( SP_Message * ) outList->takeItem( SP_ArrayList::LAST_INDEX );

		int index = msg->getToList()->find( sid );
		if( index >= 0 ) msg->getToList()->take( index );
		msg->getFailure()->add( sid );

		if( msg->getToList()->getCount() <= 0 ) {
			doCompletion( eventArg, msg );
		}
	}

	// remove session from SessionManager, onResponse will ignore this session
	eventArg->getSessionManager()->remove( sid.mKey, sid.mSeq );

	eventArg->getInputResultQueue()->push( new SP_SimpleTask( timeout, session, 1 ) );
}

void SP_EventHelper :: timeout( void * arg )
{
	SP_Session * session = ( SP_Session * )arg;
	SP_EventArg * eventArg = (SP_EventArg*)session->getArg();

	SP_Sid_t sid = session->getSid();

	SP_Response * response = new SP_Response( sid );
	session->getHandler()->timeout( response );
	msgqueue_push( (struct event_msgqueue*)eventArg->getResponseQueue(), response );

	sp_syslog( LOG_WARNING, "session(%d.%d) timeout, r %d(%d), w %d(%d), i %d, o %d, s %d(%d)",
			sid.mKey, sid.mSeq, session->getTotalRead(), session->getReading(),
			session->getTotalWrite(), session->getWriting(),
			session->getInBuffer()->getSize(), session->getOutList()->getCount(),
			eventArg->getSessionManager()->getCount(), eventArg->getSessionManager()->getFreeCount() );

	// onResponse will ignore this session, so it's safe to destroy session here
	session->getHandler()->close();
	sp_close( EVENT_FD( session->getWriteEvent() ) );
	delete session;
}

void SP_EventHelper :: doClose( SP_Session * session )
{
	SP_EventArg * eventArg = (SP_EventArg*)session->getArg();

	event_del( session->getWriteEvent() );
	event_del( session->getReadEvent() );

	SP_Sid_t sid = session->getSid();

	eventArg->getSessionManager()->remove( sid.mKey, sid.mSeq );

	eventArg->getInputResultQueue()->push( new SP_SimpleTask( myclose, session, 1 ) );
}

void SP_EventHelper :: myclose( void * arg )
{
	SP_Session * session = ( SP_Session * )arg;
	SP_EventArg * eventArg = (SP_EventArg*)session->getArg();
	SP_Sid_t sid = session->getSid();

	sp_syslog( LOG_DEBUG, "session(%d.%d) close, r %d(%d), w %d(%d), i %d, o %d, s %d(%d)",
			sid.mKey, sid.mSeq, session->getTotalRead(), session->getReading(),
			session->getTotalWrite(), session->getWriting(),
			session->getInBuffer()->getSize(), session->getOutList()->getCount(),
			eventArg->getSessionManager()->getCount(), eventArg->getSessionManager()->getFreeCount() );

	session->getHandler()->close();
	sp_close( EVENT_FD( session->getWriteEvent() ) );
	delete session;
}

void SP_EventHelper :: doStart( SP_Session * session )
{
	session->setRunning( 1 );
	SP_EventArg * eventArg = (SP_EventArg*)session->getArg();
	eventArg->getInputResultQueue()->push( new SP_SimpleTask( start, session, 1 ) );
}

void SP_EventHelper :: start( void * arg )
{
	SP_Session * session = ( SP_Session * )arg;
	SP_EventArg * eventArg = (SP_EventArg*)session->getArg();

	SP_IOChannel * ioChannel = session->getIOChannel();

	int initRet = ioChannel->init( EVENT_FD( session->getWriteEvent() ) );

	// always call SP_Handler::start
	SP_Response * response = new SP_Response( session->getSid() );
	int startRet = session->getHandler()->start( session->getRequest(), response );

	int status = SP_Session::eWouldExit;

	if( 0 == initRet ) {
		if( 0 == startRet ) status = SP_Session::eNormal;
	} else {
		delete response;
		// make an empty response
		response = new SP_Response( session->getSid() );
	}

	session->setStatus( status );
	session->setRunning( 0 );
	msgqueue_push( (struct event_msgqueue*)eventArg->getResponseQueue(), response );
}

void SP_EventHelper :: doCompletion( SP_EventArg * eventArg, SP_Message * msg )
{
	eventArg->getOutputResultQueue()->push( msg );
}