///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoDbThread.cc  
// ---------------                                                     
// Cego db 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: CegoDbThread
// 
// Description: This class implements the functions of a single database thread.
//              It serves a database session dispatched from the database thread pool
//              and handles subsequent requests for this session.
//              Database requests can be request from a native client but also distributed
//              requests from a different cego database node.
//  
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

// INCLUDES
#include <lfc/Exception.h>
#include <lfc/Sleeper.h>
#include <lfc/ThreadLock.h>

#include "CegoXMLdef.h"
#include "CegoDefs.h"
#include "CegoDbThread.h"
#include "CegoDbThreadPool.h"

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

CegoDbThread::CegoDbThread() : Thread()
{
}

CegoDbThread::CegoDbThread(CegoDbThreadPool *pPool, CegoDatabaseManager *pDBMng, CegoDbHandler::ProtocolType protType) : Thread()
{
    _pPool = pPool;
    _pDBMng = pDBMng;
    _protType = protType;
    _modId = _pDBMng->getModId("CegoDbThread");
    _errorCode = 0;

}

CegoDbThread::~CegoDbThread()
{
    delete _pPA;
    delete _pTabMng;
}

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

long CegoDbThread::allocatedSortArea() const
{
    if ( _pTabMng )
	return _pTabMng->getAllocatedSortArea();
    return 0;
}

void CegoDbThread::abortSession()
{
    _pTabMng->abort();
}

int CegoDbThread::getErrorCode() const
{
    return _errorCode;
}

void CegoDbThread::addReloadRequest(int tabSetId)
{
    _pPool->P(_idx);
    _loadList.Insert(tabSetId);
    _pPool->V(_idx);
}

void CegoDbThread::checkReloadRequest()
{   

    _pPool->P(_idx);
    int *pTS = _loadList.First();
    if ( pTS )
    {
	int ts = *pTS;
	_loadList.Remove(*pTS);
	_pPool->V(_idx);

	_pTabMng->disableAuth();

	try
	{

	    loadObjects(ts);
	}
	catch ( Exception e )
	{

	    Chain msg;
	    e.pop(msg);
	    _pDBMng->log(_modId, Logger::LOGERR, Chain("Thread ") + Chain(_idx) + Chain(" reload error : ") + msg);	   	    
	}

	_pTabMng->enableAuth();

    }
    else
    {
	_pPool->V(_idx);
    }
}

void* CegoDbThread::job(void* arg)
{

    try
    {

	_idx = *(long*)arg;
	
	_pTabMng = new CegoDistManager(_pDBMng);
	_pPA = new CegoAction(_pTabMng, _pPool);
	
	_pTim = new NanoTimer();

	_pTabMng->setPoolSyncInfo(_pPool, _idx);
	_pTabMng->setThreadId(getTid());
	_pPool->setTid(_idx, getTid());
	_pPool->setThreadState(_idx, CegoDbThreadPool::READY);
	
	while ( ! _pPool->isTerminated() )
	{
	    try
	    {
		_pTim->reset();
		_pTim->start();
		
		_pRequest = _pPool->nextRequest();
		
		if ( _pRequest ) 
		{
		    
#ifdef CGDEBUG
		    _pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Serving service request"));
#endif

		    _pPool->setState(_idx, CegoDbThreadPool::CONNECTED);
		    _pPool->incNumRequest(_idx);
		    
		    _pDBMng->increaseActiveDbThread();
		    
		    CegoDistDbHandler *pSH = 0;
		    try {
			
			pSH = new CegoDistDbHandler(_pRequest, _protType, _pDBMng); 
			serveSession(pSH);
			delete pSH;
		    }
		    catch ( Exception e)
		    {
			if ( pSH )
			    delete pSH;
			
			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(" : DB session aborted : ") + exep);
			
		    }
		    
		    // in any case, rollback any open transactions
		    _pTabMng->rollbackDistTransaction(_pPA->getTableSet());
		    
		    _pDBMng->decreaseActiveDbThread();
		    
#ifdef CGDEBUG
		    _pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Service request finished"));
#endif

		    _pPool->setState(_idx, CegoDbThreadPool::READY);
		    
		    if ( _pTabMng->isAborted() )
		    {
			_pDBMng->log(_modId, Logger::NOTICE,  Chain("Thread ") + Chain(_idx) + Chain(" : Abort catched, proceed with session"));
			_pTabMng->proceed();
		    }

		    // reset append and autocommit mode
		    _pTabMng->setAppend(false);
		    _pTabMng->setAutoCommit(true);
		   
		    delete _pRequest;
		}
		else
		{
		    Sleeper s;
		    s.nanoSleep(NETMNG_QUEUE_DELAY);

		}

		checkReloadRequest();

		_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);
	    }
	}
    }
    catch ( Exception e )
    {
	Chain msg;
	e.pop(msg);	
	_pDBMng->log(_modId, Logger::LOGERR, Chain("Thread ") + Chain(_idx) + Chain(" : ") + msg);
	_errorCode = 1;
    }
    return 0;
}

void CegoDbThread::loadObjects(int tabSetId)
{

    Chain tableSet = _pDBMng->getTabSetName(tabSetId);

    _pPA->setTableSet(tableSet);
    
    ListT<Chain> procList;
    _pTabMng->getObjectList(tabSetId, CegoObject::PROCEDURE, procList);
    
    Chain *pProcName = procList.First();
    while ( pProcName )
    {	
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Compiling procedure ") + *pProcName);
#endif
	
	CegoProcObject po;
	_pTabMng->getObject(tabSetId, *pProcName, CegoObject::PROCEDURE, po);
	
	Chain loadString = Chain("load ") + po.getProcText();
	
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Loading <<<  ") + loadString + Chain(">>>"));
#endif

	// cout << "Loading procedure " << loadString << endl;
	
	try
	{
	    _pPA->cleanUp();
	    _pPA->setCommandChain(loadString);
	    _pPA->parse();
	    
	    CegoProcedure* pProc = _pPA->getProcedure();
	    _pTabMng->addCompProcedure(tabSetId, pProc);
	    
#ifdef CGDEBUG
	    _pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Procedure ") + *pProcName + Chain(" added"));
#endif


	}
	catch ( Exception e ) 
	{

	    Chain msg;
	    e.pop(msg);	    
	    _pDBMng->log(_modId, Logger::LOGERR, Chain("Thread ") + Chain(_idx) + Chain(" procedure load error : ") + msg);
	    _pPA->cleanUp();
	}

	
	pProcName = procList.Next();
    }
    
    ListT<Chain> viewList;
    _pTabMng->getObjectList(tabSetId, CegoObject::VIEW, viewList);
    
    Chain *pViewName = viewList.First();
    while ( pViewName )
    {		
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Compiling view ") + *pViewName);
#endif
	CegoViewObject vo;
	_pTabMng->getObject(tabSetId, *pViewName, CegoObject::VIEW, vo);
	
	Chain loadString = Chain("load ") + vo.getViewStmt();
	
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Loading <<<  ") + loadString + Chain(">>>"));
#endif
	
	
	_pPA->cleanUp();
	_pPA->setCommandChain(loadString);
	_pPA->parse();
	
	CegoSelect* pSelect = _pPA->getSelect();
	
	CegoView *pView = new CegoView(*pViewName, pSelect);
	
	_pTabMng->addCompView(tabSetId, pView);
	
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : View ") + *pViewName + Chain(" added"));	
#endif
	pViewName = viewList.Next();
    }

}

void CegoDbThread::unloadObjects(int tabSetId)
{
    try
    {
	_pTabMng->getTransactionManager()->release(tabSetId);
	_pTabMng->removeAllComp(tabSetId);
    }
    catch ( Exception e )
    {
	throw e;
    }
}

void CegoDbThread::invalidateObject(int tabSetId, const Chain& objName, CegoObject::ObjectType type)
{
    
    if ( type == CegoObject::VIEW )
    {
	_pTabMng->removeCompView(tabSetId, objName);
    }
    else if ( type == CegoObject::PROCEDURE )
    {
	_pTabMng->removeCompProcedure(tabSetId, objName);
    }
    else
    {
	throw Exception(EXLOC, "Unknown object type for invalidation");	    
    }    
}

void CegoDbThread::serveSession(CegoDistDbHandler *pSH)
{       

#ifdef CGDEBUG
    _pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Serving session"));
#endif

    if ( pSH->acceptSession() )
    {
	bool isTraceOn;
	Chain msg;
	if ( _pDBMng->checkUser(pSH->getUser(), pSH->getPassword(), msg, isTraceOn) == false )
	{
	    pSH->sendError(msg);
	    return;
	}
	else
	{

	    if ( _pDBMng->tableSetExists(pSH->getTableSet()) == false )
	    {
		Chain msg = Chain("Unknown tableset ") + pSH->getTableSet();
		pSH->sendError(msg);
		return;	    
	    }
		     
	    _pTabMng->setActiveUser(pSH->getTableSet(), pSH->getUser(), pSH->getPassword());

	    msg = Chain("Access granted");
	    pSH->sendSessionConfirm(msg, _idx, XML_DBPRODNAME_VALUE, XML_DBPRODVERSION_VALUE);
	    
#ifdef CGDEBUG
	    _pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Accepted session for tableset ") + pSH->getTableSet());
#endif
	    
	}
	
	
	_pPA->setTableSet(pSH->getTableSet());
	
	bool isTerminated = false;

	while ( isTerminated == false && _pPool->isTerminated() == false )
	{
	    
	    CegoDbHandler::RequestType reqType;
	    reqType = pSH->acceptRequest();

	    _pTim->stop();
	    _pPool->addThreadIdle(_idx, _pTim->getSum());
	    _pTim->reset();
	    _pTim->start();
	    
	    if ( reqType != CegoDbHandler::REQTIMEOUT )
	    {
		Chain runState = _pDBMng->getTableSetRunState(pSH->getTableSet());
		if ( runState != Chain(XML_ONLINE_VALUE) && runState != Chain(XML_BACKUP_VALUE) )
		{
		    Chain msg = Chain("Tableset " + pSH->getTableSet() + " not online ( run state is " + runState + ")");
		    pSH->sendError(msg);
		}
		else
		{

		    if ( isTraceOn )
		    {
			_pDBMng->incUserQuery(pSH->getUser());
		    }

		    _pPool->incNumQueryRequest(_idx);
		    		    
		    _pPool->setState(_idx, CegoDbThreadPool::BUSY);


		    _pTim->stop();
		    _pPool->addThreadIdle(_idx, _pTim->getSum());
		    
		    isTerminated = serveRequest(pSH, reqType);
		    
		    _pTim->reset();
		    _pTim->start();

		    _pPool->setState(_idx, CegoDbThreadPool::CONNECTED);	    
		}
	    }
	    else
	    {


#ifdef CGDEBUG
		_pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Client request timeout occured, waiting ..."));
#endif
	    }

	    checkReloadRequest();


	}
    }
}

bool CegoDbThread::serveRequest(CegoDistDbHandler *pSH, CegoDbHandler::RequestType reqType)
{
    bool isTerminated=false;   	       

    // at the beginning of the request, in any case we reset abort status
    _pTabMng->proceed();

    try {
	
	switch ( reqType ) 
	{
	    
	case CegoDbHandler::DBPRODINFO:
	{
	    pSH->sendProdInfo();
	    break;
	}
	case CegoDbHandler::QUERYABORT:
	{
	    long idx;

	    idx = pSH->getTid();
	    _pPool->abortThread(idx);
	    pSH->sendResponse(Chain("Query aborted"));
	    break;
	}
	case CegoDbHandler::QUERY:
	{
	    
	    _lastAction = pSH->getQueryArg();
	    
#ifdef CGDEBUG
	    _pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Query Arg = <<<") + _lastAction + Chain(">>>"));
#endif
	    
	    _pPA->cleanUp();
	    _pPA->setCommandChain( _lastAction );
	    _pPA->setDbHandle(pSH);
	    
	    _pPA->parse();

#ifdef CGDEBUG
	    _pDBMng->log(_modId, Logger::DEBUG, Chain("Thread ") + Chain(_idx) + Chain(" : Query request done"));
#endif
	    
	    break;
	}
	case CegoDbHandler::INSERT:
	{
	    Chain tableSet;
	    Chain tableName;
	    ListT<CegoField> fl;
	    
	    pSH->getInsertArg(tableSet, tableName,fl);
	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    
	    CegoTableObject oe;
	    _pTabMng->getObject(tabSetId, tableName, CegoObject::TABLE, oe);
	    _pTabMng->insertLocalDataTable(oe, fl);
	    
	    pSH->sendResponse(Chain("insert ok"));
	    
	    break;
	}
	case CegoDbHandler::DELETETABLE:
	{
	    
	    Chain tableSet;
	    Chain tableName;
	    CegoPredDesc *pPred = 0;
	    
	    pSH->getDeleteArg(tableSet, tableName, pPred, _pTabMng);
	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    
	    CegoTableObject oe;
	    _pTabMng->getObject(tabSetId, tableName, CegoObject::TABLE, oe);
	    
	    long delCount = _pTabMng->deleteLocalDataTable(oe, pPred);
	    	    
	    pSH->sendResponse(Chain("delete ok"), delCount);
	    
	    break;
	}
	case CegoDbHandler::UPDATE:
	{
	    
	    Chain tableSet;
	    Chain tableName;
	    CegoPredDesc *pPred = 0;
	    ListT<CegoField> updSchema;
	    ListT<CegoExpr*> exprList;
	    
	    pSH->getUpdateArg(tableSet, tableName, updSchema, exprList, pPred, _pTabMng);
	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    
	    CegoTableObject oe;
	    _pTabMng->getObject(tabSetId, tableName, CegoObject::TABLE, oe);
	    
	    long updCount = _pTabMng->updateLocalDataTable(oe, pPred, updSchema, exprList);
	    
	    pSH->sendResponse(Chain("update ok"), updCount);
	    
	    break;
	}	
	case CegoDbHandler::CREATETABLE:
	{	    
	    Chain tableSet;
	    Chain tableName;
	    ListT<CegoField> fl;
	    ListT<CegoField> idxList;
	    
	    pSH->getCreateTableArg(tableSet, tableName, fl, idxList);
	    int tabSetId = _pDBMng->getTabSetId(tableSet);	    
	    _pTabMng->createLocalDataTable(tabSetId, tableName, CegoObject::TABLE, fl, idxList);	  	    

	    _pDBMng->addObject(tabSetId, tableName, CegoObject::TABLE); 
	    if ( ! idxList.isEmpty() )
	    {
		Chain idxName = tableName + Chain(TABMNG_PIDX_SUFFIX);	
		_pDBMng->addObject(tabSetId, idxName, CegoObject::INDEX);	
	    }

	    pSH->sendResponse(Chain("create ok"));	    
	    break;	    
	}
	case CegoDbHandler::CREATEVIEW:
	{	    
	    Chain tableSet;
	    Chain viewName;
	    Chain viewText;
	    ListT<CegoField> fl;
	    
	    pSH->getCreateViewArg(tableSet, viewName, fl, viewText);
	    int tabSetId = _pDBMng->getTabSetId(tableSet);	    
	    _pTabMng->createLocalView(tabSetId, viewName, fl, viewText);	  	    

	    _pDBMng->addObject(tabSetId, viewName, CegoObject::VIEW);

	    pSH->sendResponse(Chain("create view ok"));	    
	    break;	    
	}
	case CegoDbHandler::CREATEPROCEDURE:
	{	    
	    Chain tableSet;
	    Chain procName;
	    Chain procText;
	    
	    pSH->getCreateProcedureArg(tableSet, procName, procText);
	    int tabSetId = _pDBMng->getTabSetId(tableSet);	    
	    _pTabMng->createLocalProc(tabSetId, procName, procText);	  	    

	    _pDBMng->addObject(tabSetId, procName, CegoObject::PROCEDURE);

	    pSH->sendResponse(Chain("create procedure ok"));	    
	    break;	    
	}
	case CegoDbHandler::CREATECHECK:
	{	    
	    Chain tableSet;
	    Chain checkName;
	    Chain tableName;
	    CegoPredDesc *pPred;

	    pSH->getCreateCheckArg(tableSet, checkName, tableName, pPred, _pTabMng);
	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);	    
	    _pTabMng->createLocalCheck(tabSetId, checkName, tableName, pPred);	  	    

	    _pDBMng->addObject(tabSetId, checkName, CegoObject::CHECK);

	    pSH->sendResponse(Chain("create check ok"));	    
	    break;	    
	}

	case CegoDbHandler::ALTERTABLE:
	{	    
	    Chain tableSet;
	    Chain tableName;
	    ListT<CegoAlterDesc> alterList;
	    
	    pSH->getAlterTableArg(tableSet, tableName, alterList);	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);	    
	    _pTabMng->alterDataTable(tabSetId, tableName, CegoObject::TABLE, alterList);	    
	    pSH->sendResponse(Chain("alter ok"));	    
	    break;
	}
	case CegoDbHandler::DROPOBJECT:
	{	    
	    Chain tableSet;
	    CegoObject::ObjectType objType; 
	    Chain objName;
	    
	    pSH->getDropTableArg(tableSet, objName, objType);	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    _pTabMng->dropLocalObject(tabSetId, objName, objType);
	    pSH->sendResponse(Chain("Drop ok"));	    
	    break;	    
	}
	case CegoDbHandler::CREATEINDEX:
	{	    
	    Chain tableSet;
	    Chain indexName;
	    Chain tableName;
	    ListT<CegoField> idxList;
	    CegoObject::ObjectType type;	    
	    pSH->getCreateIndexArg(tableSet, indexName, tableName, idxList, type);	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    _pTabMng->createLocalIndexTable(tabSetId, indexName, tableName, type, idxList);

	    _pDBMng->addObject(tabSetId, indexName, type);

	    pSH->sendResponse(Chain("create ok"));	    
	    break;	    
	}
	case CegoDbHandler::OBJECTINFO:
	{
	    int tabSetId;
	    Chain objName;
	    CegoObject::ObjectType objType;
	    
	    pSH->getObjectInfoArg(tabSetId, objName, objType);
	    
	    if ( _pDBMng->objectExists(tabSetId, objName, objType) == false )
		pSH->sendError(Chain("Object does not exist"));
	    else
	    {
	    
		switch ( objType )
		{
		case CegoObject::SYSTEM:
		case CegoObject::TABLE:
		case CegoObject::PINDEX:
		case CegoObject::UINDEX:
		case CegoObject::INDEX:
		{   
		    CegoTableObject to;
		    _pTabMng->getLocalObject(tabSetId, objName, objType, to);
		    pSH->sendObjInfo(to);
		    break;
		}
		case CegoObject::BTREE:
		case CegoObject::PBTREE:
		case CegoObject::UBTREE:
		{   
		    CegoBTreeObject bto;
		    _pTabMng->getLocalObject(tabSetId, objName, objType, bto);
		    pSH->sendObjInfo(bto);
		    break;
		}
	
		case CegoObject::VIEW:
		{
		    CegoViewObject vo;
		    _pTabMng->getLocalObject(tabSetId, objName, objType, vo);
		    pSH->sendObjInfo(vo);
		    break;
		}
		case CegoObject::PROCEDURE:
		{
		    CegoProcObject po;
		    _pTabMng->getLocalObject(tabSetId, objName, objType, po);
		    pSH->sendObjInfo(po);
		    break;
		}
		case CegoObject::FKEY:
		{
		    CegoKeyObject ko;
		    _pTabMng->getLocalObject(tabSetId, objName, objType, ko);
		    pSH->sendObjInfo(ko);
		    break;
		}
		case CegoObject::CHECK:
		{
		    CegoCheckObject co;
		    _pTabMng->getLocalObject(tabSetId, objName, objType, co);
		    pSH->sendObjInfo(co);
		    break;
		}
		case CegoObject::JOIN:
		case CegoObject::RBSEG:
		case CegoObject::UNDEFINED:
		{
		    throw Exception(EXLOC, "Objectinfo not supported for this object type");
		    break;
		}
		}
	    }
	    break;
	}
	case CegoDbHandler::GETTABLE:
	{
	    int tabSetId;
	    Chain tableName;
	    CegoObject::ObjectType type;
	    
	    pSH->getGetTableArg(tabSetId, tableName, type);
	    
	    CegoTableObject oe;
	    _pTabMng->getLocalObject(tabSetId, tableName, type, oe);
	    
	    ListT<CegoField> schema = oe.getSchema();		
	    
	    pSH->collectSchema(schema);
	    
	    CegoObjectCursor *pOC = _pTabMng->getObjectCursor(tabSetId, tableName, tableName, type);
	    
	    int tupleCount=0;
	    
	    CegoDataPointer dp;
	    bool moreTuple = _pTabMng->getFirstTuple(pOC, schema, dp);
	    
	    while ( moreTuple )
	    {
		pSH->collectData(schema);
		tupleCount++;
		
		if ( tupleCount == NETMNG_MAXTUPLECOUNT )
		{
		    pSH->sendCollectedData();
		    tupleCount=0;
		}
		moreTuple = _pTabMng->getNextTuple(pOC, schema, dp);
	    }
	    
	    if ( tupleCount > 0)
	    {
		pSH->sendCollectedData();
	    }
	    
	    pSH->sendFinishData();
	 	    
	    break;
	}
	case CegoDbHandler::GETOBJLIST:
	{
	    int tabSetId;
	    CegoObject::ObjectType type;
	    
	    pSH->getGetObjectListArg(tabSetId, type);
	    
	    ListT<Chain> objList;
	    _pTabMng->getObjectList(tabSetId, type, objList);
	    
	    pSH->sendObjList(objList);
	    
	    break;
	    
	}
	case CegoDbHandler::GETOBJLISTBYTABLE:
	{
	    Chain tableSet;
	    Chain tableName;
	    
	    pSH->getGetObjectByTableListArg(tableSet, tableName);
	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    
	    
	    ListT<CegoTableObject> idxList;
	    ListT<CegoBTreeObject> btreeList;
	    ListT<CegoKeyObject> keyList;
	    ListT<CegoCheckObject> checkList;

	    _pTabMng->getObjectListByTable(tabSetId, tableName, idxList, btreeList, keyList, checkList);
	    
	    pSH->sendObjByTableList(idxList, keyList, checkList);
	    
	    break;
	    
	}	
	case CegoDbHandler::OBJRENAME:
	{
	    Chain tableSet;
	    CegoObject::ObjectType type;
	    Chain objName;
	    Chain newObjName;
	    
	    pSH->getRenameArg(tableSet, objName, type, newObjName);
	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    
	    _pTabMng->renameObject(tabSetId, objName, type, newObjName);
	    
	    pSH->sendResponse(Chain("rename ok"));
	    
	    break;
	    
	}
	case CegoDbHandler::REORGOBJ:
	{
	    Chain tableSet;
	    CegoObject::ObjectType type;
	    Chain objName;
	    Chain newObjName;
	    
	    pSH->getReorgArg(tableSet, objName, type);
	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    
	    _pTabMng->reorgObject(tabSetId, objName, type);
	    
	    pSH->sendResponse(Chain("reorg ok"));
	    
	    break;
	    
	}

	case CegoDbHandler::SYNC:
	{
	    Chain tableSet;
	    Chain escCmd;
	    int timeout;
	    
	    pSH->getSyncArg(tableSet, escCmd, timeout);
	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    
	    _pTabMng->writeCheckPoint(tabSetId, true, true, escCmd, timeout);
	    
	    pSH->sendResponse(Chain("sync ok"));
	    
	    break;
	    
	}
	case CegoDbHandler::GETPAGECOUNT:
	{
	    Chain tableSet;
	    Chain objName;
	    CegoObject::ObjectType type;
	    
	    pSH->getPageCountArg(tableSet, objName, type);
	    
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    
	    int pageCount = 0;
	    _pTabMng->getPageCount(tabSetId, objName, type);
	    
	    pSH->sendPageCount(pageCount);
	    
	    break;
	    
	}
	case CegoDbHandler::PUTBLOB:
	{

	    _lastAction = Chain("put blob");

	    Chain tableSet;
	    long blobSize;
	    
	    pSH->getPutBlobArg(tableSet, blobSize);

	    int tabSetId = _pDBMng->getTabSetId(tableSet);

	    NetHandler *pN = pSH->getNetHandler();

	    int fileId = 0;
	    int pageId = 0;

	    CegoBufferPage bp;

	    try
	    {

		_pTabMng->getNewFilePage(bp, tabSetId, CegoObject::TABLE);
		bp.initPage(CegoBufferPage::BLOB);
		
		fileId = bp.getFileId();
		pageId = bp.getPageId();
		

#ifdef CGDEBUG
		_pDBMng->log(_modId, Logger::DEBUG, Chain(" Thread ") + Chain(_idx) + Chain(" : Sending blob info [") + Chain(fileId) + Chain(",") + Chain(pageId) + Chain("]"));
#endif
		pSH->sendBlobInfo(fileId, pageId);
		
		int freeInPage = bp.getChunkLen();
		char* pagePtr = bp.getChunkEntry();
		
		memcpy(pagePtr, &blobSize, sizeof(long));
		pagePtr += sizeof(long);
		freeInPage -= sizeof(long);
		
		char *bufPtr = 0; 
		int availFromBuf = 0;
		
		long recvSize = 0;
		while ( recvSize < blobSize )
		{		    
		    if ( availFromBuf == 0 )		    
		    {
			if ( pN->waitMsg(NETMNG_WAITMSG_TIMEOUT) == false )
			    throw Exception(EXLOC, "Timeout exceeded for blob chunk");
			pN->readMsg();
			pN->sendAck();
			bufPtr = pN->getMsg();
			availFromBuf = pN->getMsgSize();
		    }    
		    
		    if ( freeInPage == 0 )
		    {
			CegoBufferPage nextPage; 
			_pTabMng->getNewFilePage(nextPage, tabSetId, CegoObject::TABLE);
			nextPage.initPage(CegoBufferPage::BLOB);
			
			bp.setNextFileId(nextPage.getFileId());
			bp.setNextPageId(nextPage.getPageId());

			_pDBMng->bufferUnfix(bp, true, _pTabMng->getLockHandler());
			
			bp = nextPage;
			
			freeInPage = bp.getChunkLen();
			pagePtr = bp.getChunkEntry();
		    }
		    
		    if ( freeInPage >= availFromBuf )
		    {
			memcpy(pagePtr, bufPtr, availFromBuf);
			recvSize += availFromBuf;
			pagePtr += availFromBuf;
			freeInPage -= availFromBuf;
			availFromBuf = 0;
		    }
		    else if ( freeInPage < availFromBuf )
		    {
			memcpy(pagePtr, bufPtr, freeInPage);
			recvSize += freeInPage;
			bufPtr += freeInPage;
			availFromBuf -= freeInPage;
			freeInPage = 0;
		    }		    
		}
		_pDBMng->bufferUnfix(bp, true, _pTabMng->getLockHandler());
	    }
	    catch ( Exception e )
	    {
		if ( bp.isFixed() )
		    _pDBMng->bufferUnfix(bp, true, _pTabMng->getLockHandler());
		if ( fileId || pageId )
		{
		    _pTabMng->releaseBlob(tabSetId, fileId, pageId);
		}
	    }

#ifdef CGDEBUG
	    _pDBMng->log(_modId, Logger::DEBUG, Chain(" Thread ") + Chain(_idx) + Chain(" : Blob transfer finished"));
#endif

	    break;

	}
	case CegoDbHandler::GETBLOB:
	{

	    _lastAction = Chain("get blob");

	    Chain tableSet;
	    int fileId;
	    int pageId;
	    
	    pSH->getGetBlobArg(tableSet, fileId, pageId);

	    int tabSetId = _pDBMng->getTabSetId(tableSet);

	    NetHandler *pN = pSH->getNetHandler();	    	

	    bool isFixed = false;
	    CegoBufferPage bp;
	    try
	    {

		_pDBMng->bufferFix(bp, tabSetId, fileId, pageId, CegoBufferPool::SYNC,
						     _pTabMng->getLockHandler());

	    	
		long blobSize;
		memcpy (&blobSize, bp.getChunkEntry(), sizeof(long));
		
		pSH->sendBlobSize(blobSize);
		
		long sentByte = 0;
		
		while ( bp.isFixed() )
		{
		    pN->recvAck();
		    
		    int chunkSize;
		    if (  bp.getChunkLen() < blobSize - sentByte )
		    {		       
			chunkSize = bp.getChunkLen();
			if ( sentByte == 0 )
			    chunkSize -= sizeof(long);
		    }
		    else
		    {
			chunkSize = blobSize - sentByte;
		    }
		    
		    if ( sentByte == 0 )
		    {
			pN->setMsg(bp.getChunkEntry() + sizeof(long), chunkSize );
		    }
		    else
		    {
			pN->setMsg(bp.getChunkEntry(), chunkSize);
		    }
		    sentByte += chunkSize;

		    pN->writeMsg();
		    
		    fileId = bp.getNextFileId();
		    pageId = bp.getNextPageId();
		    
		    _pDBMng->bufferUnfix(bp, false, _pTabMng->getLockHandler());
		    
		    if ( fileId || pageId )			
		    {
			_pDBMng->bufferFix(bp, tabSetId, fileId, pageId, CegoBufferPool::SYNC,
					   _pTabMng->getLockHandler());
		    }		    
		}
	    }
	    catch ( Exception e )
	    {
		if ( bp.isFixed() )
		    _pDBMng->bufferUnfix(bp, true, _pTabMng->getLockHandler());
	    }

	    break;
	    
	}
	case CegoDbHandler::DELBLOB:
	{

	    _lastAction = Chain("del blob");

	    Chain tableSet;
	    int fileId;
	    int pageId;
	    
	    pSH->getGetBlobArg(tableSet, fileId, pageId);

	    int tabSetId = _pDBMng->getTabSetId(tableSet);

	    _pTabMng->releaseBlob(tabSetId, fileId, pageId);

	    pSH->sendResponse(Chain("Blob deleted"));
	    
	    break;
	    
	}
	case CegoDbHandler::STARTTRANSACTION:
	{
	    _lastAction = Chain("start transaction");

	    Chain tableSet;
	    pSH->getArgValue(XML_TABLESET_ATTR, tableSet);
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    _pTabMng->beginTransaction(tabSetId);
	    pSH->sendResponse(Chain("Transaction started"));	    
	    break;	    
	}
	case CegoDbHandler::COMMITTRANSACTION:
	{
	    _lastAction = Chain("commit transaction");
	    Chain tableSet;
	    pSH->getArgValue(XML_TABLESET_ATTR, tableSet);
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    _pTabMng->commitTransaction(tabSetId);
	    pSH->sendResponse(Chain("Transaction committed"));	    
	    break;	    
	}
	case CegoDbHandler::ROLLBACKTRANSACTION:
	{
	    _lastAction = Chain("rollback transaction");
	    Chain tableSet;
	    pSH->getArgValue(XML_TABLESET_ATTR, tableSet);
	    int tabSetId = _pDBMng->getTabSetId(tableSet);
	    _pTabMng->rollbackTransaction(tabSetId);
	    pSH->sendResponse(Chain("Transaction rollbacked"));	    
	    break;	    
	}
	case CegoDbHandler::GETTID:
	{
	    _lastAction = Chain("get tid");
	    Chain tableSet;
	    pSH->getArgValue(XML_TABLESET_ATTR, tableSet);
	    int tabSetId = _pDBMng->getTabSetId(tableSet);	    
	    pSH->sendTID(_pTabMng->getTID(tabSetId));
	    break;	    
	}
	case CegoDbHandler::SESSION_CLOSE:
	{
	    isTerminated=true;
	    pSH->sendResponse(Chain("Session Closed"));
	    break;
	}
	case CegoDbHandler::REQTIMEOUT:
	case CegoDbHandler::UNKNOWN:
	{
	    break;
	}
	}
    }
    catch ( Exception e)
    {
	
	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(" : DB request failed : ") + exep);
	
	if ( _pTabMng->isAborted() )
	{
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Thread ") + Chain(_idx) + Chain(" : Abort catched, proceed with session"));
	    _pTabMng->proceed();
	}
	
	pSH->sendError(e.getBaseMsg());
    }

    return isTerminated;
}

