///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoLogThread.cc  
// ----------------                                                     
// Cego log thread class implementation
//                                                         
// Design and Implementation by Bjoern Lemke
//               
// (C)opyright 2000-2010 Bjoern Lemke
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2, or (at your option)
// any later version.
// 
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; see the file COPYING.  If not, write to
// the Free Software Foundation, 59 Temple Place - Suite 330,
// Boston, MA 02111-1307, USA.
//
// IMPLEMENTATION MODULE
//
// Class: CegoLogThread
// 
// Description: 
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

// base includes
#include <lfc/Exception.h>
#include <lfc/Host.h>
#include <lfc/Sleeper.h>

// cego includes
#include "CegoLogThread.h"
#include "CegoLogThreadPool.h"
#include "CegoDefs.h"
#include "CegoXMLdef.h"

CegoLogThread::CegoLogThread() : Thread()
{
}

CegoLogThread::CegoLogThread(CegoLogThreadPool *pPool, CegoDatabaseManager *pDBMng) : Thread()
{
    _pDBMng = pDBMng;
    _pPool = pPool;
    _modId = pDBMng->getModId("CegoLogThread");
}

CegoLogThread::~CegoLogThread()  
{
}

const Chain& CegoLogThread::lastAction() const
{
    return _lastAction;
}

void* CegoLogThread::job(void* arg)
{
    _idx = *(long*)arg;

    _pTim = new NanoTimer();
    
    _pPool->setTid(_idx, getTid());
    
    while ( ! _pPool->isTerminated() )
    {	
	try
	{	    

	    _pTim->reset();
	    _pTim->start();

	    _pRequest = _pPool->nextRequest();
	    
	    if ( _pRequest ) 
	    {
		
		_pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(": Serving service request"));
		_pPool->setState(_idx, CegoLogThreadPool::CONNECTED);
		_pPool->incNumRequest(_idx);

		_pDBMng->increaseActiveLogThread();

		CegoLogHandler *pLH = 0;

		try {

		    pLH = new CegoLogHandler(_pDBMng, _pRequest);		    
		    serveSession(pLH);
		    delete pLH;
		}
		catch ( Exception e)
		{	
		    if ( pLH )
			delete pLH;


		    Chain msg;
		    Chain module;
		    int line;
		    
		    Chain exep;
		    while ( e.pop(module, line, msg) )
		    {
			exep += Chain("\n\t") + module + Chain("(") +  Chain(line) + Chain(") : ") + msg;
		    }

		    _pDBMng->log(_modId, Logger::LOGERR, Chain("Thread ") + Chain(_idx) + Chain(": Log session aborted : ") + exep);

		}

		_pDBMng->decreaseActiveLogThread();
		_pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(": service request finished"));
		_pPool->setState(_idx, CegoLogThreadPool::READY);

		delete _pRequest;
	    }
	    else
	    {
		Sleeper s;
		s.nanoSleep(NETMNG_QUEUE_DELAY);
	    }

	    _pTim->stop();
	    _pPool->addThreadIdle(_idx, _pTim->getSum());
	
	}
	catch ( Exception e)
	{
	    Chain msg;
	    e.pop(msg);
	    _pDBMng->log(_modId, Logger::LOGERR, Chain("Thread ") + Chain(_idx) + Chain(": ") + msg);
	}
    }
    return 0;
}

void CegoLogThread::serveSession(CegoLogHandler *pLH)
{   

    _pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(": serving session"));
    
    Chain tableSet;

    if ( pLH->acceptLogSession(tableSet) )
    {
	char* logEntry;
	int len;
	int logCount=0;

	File* pLogFile = 0;

	int offset=0;

	_pPool->setState(_idx, CegoLogThreadPool::BUSY);
	
	Host h;
	_pDBMng->setSecondary(tableSet, h.getName());
	
	_lastAction = Chain("Recovering tableset ") + tableSet;
	
	_pTim->stop();
	_pPool->addThreadIdle(_idx, _pTim->getSum());

	while ( pLH->receiveLogEntry(logEntry, len) )
	{

	    CegoLogRecord lr;
	    lr.decodeLSN(logEntry);
	    lr.decodeLogAction(logEntry);
	    
	    _pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(": receiving lsn ") + Chain(lr.getLSN()));

	    if ( logCount % LOGMNG_MAXLOGCOUNT == 0 || lr.getAction() == CegoLogRecord::LOGREC_SYNC )
	    {
		
		_pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(": switching logfiles on lsn ") + Chain(lr.getLSN()));
		if ( pLogFile )
		{
		    pLogFile->seek(0);
		    pLogFile->writeByte((char*)&offset, sizeof(int));
		    pLogFile->trunc(offset);
		    
		    pLogFile->close();
		    delete pLogFile;
		}
		
		ListT<Chain> archPathList;
		ListT<Chain> archIdList;
		_pDBMng->getArchLogInfo(tableSet, archIdList, archPathList);
		
		Chain *pArchPath = archPathList.First();
		    
		if ( pArchPath )
		{
		    Chain logFile = *pArchPath + Chain("/") + _pDBMng->getArchiveLogName(tableSet, lr.getLSN());

		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Thread ") + Chain(_idx) + Chain(": Writing to next logfile ") + logFile);

		    pLogFile=new File(logFile);
		    pLogFile->open(File::WRITE);
		    offset=0;
		    pLogFile->writeByte((char*)&offset, sizeof(int));

		    logCount=0;
		    pLH->sendAck();
		}
		else
		{
		    _pDBMng->log(_modId, Logger::LOGERR, Chain("Thread ") + Chain(_idx) + Chain(": cannot get archive path"));
		    pLH->sendNack();
		    _pPool->setState(_idx, CegoLogThreadPool::CONNECTED);
		    return;
		}
	    }
	    else
	    {
		pLH->sendAck();
	    }

	    
	    pLogFile->writeByte((char*)&len, sizeof(int));
	    pLogFile->writeByte(logEntry, len);
	    logCount++;
	    offset = offset + len + sizeof(int);

	    // writing logentry
	}
	
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(": Closing logfile ") + pLogFile->getFileName());

	if ( pLogFile )
	{
	    pLogFile->seek(0);
	    pLogFile->writeByte((char*)&offset, sizeof(int));
	    pLogFile->trunc(offset);
	    
	    pLogFile->close();
	    delete pLogFile;
	}

	_pTim->reset();
	_pTim->start();
	
	_pPool->setState(_idx, CegoLogThreadPool::CONNECTED);
	
    }

    _pTim->stop();
    _pPool->addThreadIdle(_idx, _pTim->getSum());

    _pTim->reset();
    _pTim->start();


}
