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

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>

#include "spporting.hpp"

#include "spiocplfserver.hpp"

#include "spiocpevent.hpp"
#include "spwin32iocp.hpp"
#include "spthreadpool.hpp"
#include "sphandler.hpp"
#include "spexecutor.hpp"
#include "sputils.hpp"
#include "spioutils.hpp"
#include "spiochannel.hpp"

SP_IocpLFServer :: SP_IocpLFServer( const char * bindIP, int port, SP_HandlerFactory * handlerFactory )
{
	snprintf( mBindIP, sizeof( mBindIP ), "%s", bindIP );
	mPort = port;

	mIsShutdown = 0;
	mIsRunning = 0;

	mEventArg = new SP_IocpEventArg( 600 );

	mMaxThreads = 4;

	mAcceptArg = (SP_IocpAcceptArg_t*)malloc( sizeof( SP_IocpAcceptArg_t ) );
	memset( mAcceptArg, 0, sizeof( SP_IocpAcceptArg_t ) );
	mAcceptArg->mMaxConnections = 256;
	mAcceptArg->mReqQueueSize = 128;
	mAcceptArg->mRefusedMsg = strdup( "System busy, try again later." );
	mAcceptArg->mHandlerFactory = handlerFactory;

	mAcceptArg->mEventArg = mEventArg;

	mThreadPool = NULL;

	mCompletionHandler = NULL;

	sp_thread_mutex_init( &mMutex, NULL );

	mCompletionPort = mEventArg->getCompletionPort();
}

SP_IocpLFServer :: ~SP_IocpLFServer()
{
	shutdown();

	for( ; mIsRunning; ) sleep( 1 );

	if( NULL != mThreadPool ) delete mThreadPool;
	mThreadPool = NULL;

	if( NULL != mCompletionHandler ) delete mCompletionHandler;
	mCompletionHandler = NULL;

	delete mAcceptArg->mHandlerFactory;
	free( mAcceptArg->mRefusedMsg );

	free( mAcceptArg );
	mAcceptArg = NULL;

	delete mEventArg;
	mEventArg = NULL;

	sp_thread_mutex_destroy( &mMutex );

	mCompletionPort = NULL;
}

void SP_IocpLFServer :: setTimeout( int timeout )
{
	mEventArg->setTimeout( timeout );
}

void SP_IocpLFServer :: setMaxConnections( int maxConnections )
{
	mAcceptArg->mMaxConnections = maxConnections > 0 ?
			maxConnections : mAcceptArg->mMaxConnections;
}

void SP_IocpLFServer :: setMaxThreads( int maxThreads )
{
	mMaxThreads = maxThreads > 0 ? maxThreads : mMaxThreads;
}

void SP_IocpLFServer :: setReqQueueSize( int reqQueueSize, const char * refusedMsg )
{
	mAcceptArg->mReqQueueSize = reqQueueSize > 0 ?
			reqQueueSize : mAcceptArg->mReqQueueSize;

	if( NULL != mAcceptArg->mRefusedMsg ) free( mAcceptArg->mRefusedMsg );
	mAcceptArg->mRefusedMsg = strdup( refusedMsg );
}

void SP_IocpLFServer :: setIOChannelFactory( SP_IOChannelFactory * ioChannelFactory )
{
	mAcceptArg->mIOChannelFactory = ioChannelFactory;
}

void SP_IocpLFServer :: shutdown()
{
	mIsShutdown = 1;
	
	if( NULL != mCompletionPort ) {
		PostQueuedCompletionStatus( mCompletionPort, 0, 0, 0 );
	}
}

int SP_IocpLFServer :: isRunning()
{
	return mIsRunning;
}

void SP_IocpLFServer :: sigHandler( int, short, void * arg )
{
	SP_IocpLFServer * server = (SP_IocpLFServer*)arg;
	server->shutdown();
}

void SP_IocpLFServer :: lfHandler( void * arg )
{
	SP_IocpLFServer * server = (SP_IocpLFServer*)arg;

	for( ; 0 == server->mIsShutdown; ) {
		server->handleOneEvent();
	}

	server->mIsRunning = 0;
}

void SP_IocpLFServer :: handleOneEvent()
{
	SP_Task * task = NULL;
	SP_Message * msg = NULL;

	sp_thread_mutex_lock( &mMutex );

	for( ; 0 == mIsShutdown && NULL == task && NULL == msg; ) {
		if( mEventArg->getInputResultQueue()->getLength() > 0 ) {
			task = (SP_Task*)mEventArg->getInputResultQueue()->pop();
		} else if( mEventArg->getOutputResultQueue()->getLength() > 0 ) {
			msg = (SP_Message*)mEventArg->getOutputResultQueue()->pop();
		}

		if( NULL == task && NULL == msg ) {
			SP_IocpEventCallback::eventLoop( mEventArg, mAcceptArg );
		}
	}

	sp_thread_mutex_unlock( &mMutex );

	if( NULL != task ) task->run();

	if( NULL != msg ) mCompletionHandler->completionMessage( msg );
}

sp_thread_result_t SP_THREAD_CALL SP_IocpLFServer :: acceptThread( void * arg )
{
	DWORD recvBytes = 0;

	SP_IocpAcceptArg_t * acceptArg = (SP_IocpAcceptArg_t*)arg;
	
	for( ; ; ) {
		acceptArg->mClientSocket = (HANDLE)WSASocket( AF_INET, SOCK_STREAM,
				IPPROTO_IP, NULL, 0, WSA_FLAG_OVERLAPPED );
		if( INVALID_SOCKET == (int)acceptArg->mClientSocket ) {
			sp_syslog( LOG_ERR, "WSASocket fail, errno %d", WSAGetLastError() );
			Sleep( 50 );
			continue;
		}

		SP_IOUtils::setNonblock( (int)acceptArg->mClientSocket );
		memset( &( acceptArg->mOverlapped ), 0, sizeof( OVERLAPPED ) );

		BOOL ret = AcceptEx( (SOCKET)acceptArg->mListenSocket, (SOCKET)acceptArg->mClientSocket,
				acceptArg->mBuffer,	0, sizeof(struct sockaddr_in) + 16, sizeof(struct sockaddr_in) + 16,
				&recvBytes, &( acceptArg->mOverlapped ) );

		int lastError = WSAGetLastError();
		if( FALSE == ret && (ERROR_IO_PENDING != lastError) ) {
			sp_syslog( LOG_ERR, "AcceptEx() fail, errno %d", lastError );
			closesocket( (int)acceptArg->mClientSocket );
			if( WSAENOBUFS == lastError ) Sleep( 50 );
		} else {
			WaitForSingleObject( acceptArg->mAcceptEvent, INFINITE );
			ResetEvent( acceptArg->mAcceptEvent );
		}
	}

	return 0;
}

int SP_IocpLFServer :: run()
{
	int ret = 0;
	int listenFD = -1;

	ret = SP_IOUtils::tcpListen( mBindIP, mPort, &listenFD, 0 );

	if( 0 == ret ) {
		mCompletionHandler = mAcceptArg->mHandlerFactory->createCompletionHandler();

		SP_IocpMsgQueue * msgQueue = new SP_IocpMsgQueue( mEventArg->getCompletionPort(),
			SP_IocpEventCallback::eKeyMsgQueue, SP_IocpEventCallback::onResponse, mEventArg );
		mEventArg->setResponseQueue( msgQueue );
		mEventArg->loadDisconnectEx( listenFD );

		mAcceptArg->mAcceptEvent = CreateEvent( NULL, TRUE, FALSE, NULL );
		mAcceptArg->mListenSocket = (HANDLE)listenFD;

		if( NULL == CreateIoCompletionPort( mAcceptArg->mListenSocket,
				mEventArg->getCompletionPort(), SP_IocpEventCallback::eKeyAccept, 0 ) ) {
			sp_syslog( LOG_ERR, "CreateIoCompletionPort fail, errno %d", WSAGetLastError() );
			return -1;		
		}

		if( NULL == mAcceptArg->mIOChannelFactory ) {
			mAcceptArg->mIOChannelFactory = new SP_DefaultIOChannelFactory();
		}

		sp_thread_t thread;
		ret = sp_thread_create( &thread, NULL, acceptThread, mAcceptArg );
		if( 0 == ret ) {
			sp_syslog( LOG_NOTICE, "Thread #%ld has been created to accept socket", thread );
		} else {
			sp_syslog( LOG_WARNING, "Unable to create a thread to accept socket, %s", strerror( errno ) );
			return -1;
		}

		mIsRunning = 1;

		mThreadPool = new SP_ThreadPool( mMaxThreads );
		for( int i = 0; i < mMaxThreads; i++ ) {
			mThreadPool->dispatch( lfHandler, this );
		}
	}

	return ret;
}

void SP_IocpLFServer :: runForever()
{
	run();
	pause();
}