///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoDatabaseManager.cc
// ----------------------
// Cego Database Manager 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: CegoDatabaseManager
// 
// Description: 
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

#include <lfc/ThreadLock.h>
#include <lfc/File.h>
#include <lfc/Host.h>
#include <lfc/Datetime.h>
#include <lfc/Sleeper.h>
#include <lfc/Net.h>
#include <lfc/NetHandler.h>

#include <lfc/XMLSuite.h>
#include <lfc/Element.h>
#include <lfc/Document.h>

#include "CegoDatabaseManager.h"
#include "CegoTypeConverter.h"
#include "CegoDefs.h"
#include "CegoXMLdef.h"
#include "CegoXMLSpace.h"

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

#ifndef _REENTRANT
#define _REENTRANT    /* basic 3-lines for threads */
#endif

static ThreadLock dbmLock("DBM");
extern bool __lockStatOn;

//////////////////
// ObjectRecord //
//////////////////

CegoDatabaseManager::ObjectRecord::ObjectRecord()
{
}

CegoDatabaseManager::ObjectRecord::ObjectRecord(const int tabSetId, const Chain& objName, CegoObject::ObjectType type)
{
    _objName = objName;
    _type = type;
    _tabSetId = tabSetId;
    _numUsed = 0;
    _mode = SHARED;
    _tid = 0;
}

CegoDatabaseManager::ObjectRecord::~ObjectRecord()
{
}

const int CegoDatabaseManager::ObjectRecord::getTabSetId() const
{
    return _tabSetId;
}

void CegoDatabaseManager::ObjectRecord::incUsed()
{
    _numUsed++;
}

void CegoDatabaseManager::ObjectRecord::decUsed()
{
    if ( _numUsed > 0 )
	_numUsed--;
}

int CegoDatabaseManager::ObjectRecord::numUsed() const
{
    return _numUsed;
}

void CegoDatabaseManager::ObjectRecord::setMode(ObjectUseMode mode)
{
    _mode = mode;
} 

CegoDatabaseManager::ObjectUseMode CegoDatabaseManager::ObjectRecord::getMode() const
{
    return _mode;
}

void CegoDatabaseManager::ObjectRecord::setTid(long tid)
{
    _tid = tid;
} 

long CegoDatabaseManager::ObjectRecord::getTid() const
{
    return _tid;
}

CegoDatabaseManager::ObjectRecord& CegoDatabaseManager::ObjectRecord::operator = ( const CegoDatabaseManager::ObjectRecord& t)
{
    _tabSetId = t._tabSetId;
    _objName = t._objName;
    _type = t._type;
    _numUsed = t._numUsed;
    _mode = t._mode;
    _tid = t._tid;
    return (*this);
}

bool CegoDatabaseManager::ObjectRecord::operator == ( const CegoDatabaseManager::ObjectRecord& r)
{
    if ( _objName == r._objName && _type == r._type)
	return true;
    return false;
}

//////////////////
// CopyRecord //
//////////////////

CegoDatabaseManager::CopyRecord::CopyRecord()
{
}

CegoDatabaseManager::CopyRecord::CopyRecord(const Chain& tableSet, const Chain& targetHost, const Chain& mediatorHost, const Chain& user, const Chain& passwd, const Chain& msg)
{
    _id = 0;
    _tableSet = tableSet;
    _targetHost = targetHost;
    _mediatorHost = mediatorHost;
    _user = user;
    _passwd = passwd;
    _msg = msg;
}

CegoDatabaseManager::CopyRecord::~CopyRecord()
{
}

const Chain& CegoDatabaseManager::CopyRecord::getTableSet() const
{
    return _tableSet;
}

const Chain& CegoDatabaseManager::CopyRecord::getTargetHost() const
{
    return _targetHost;
}

const Chain& CegoDatabaseManager::CopyRecord::getMediatorHost() const
{
    return _mediatorHost;
}

const Chain& CegoDatabaseManager::CopyRecord::getUser() const
{
    return _user;
}

const Chain& CegoDatabaseManager::CopyRecord::getPasswd() const
{
    return _passwd;
}

void CegoDatabaseManager::CopyRecord::setMsg(const Chain& msg)
{
    _msg = msg;
}

const Chain& CegoDatabaseManager::CopyRecord::getMsg() const
{
    return _msg;
}

int CegoDatabaseManager::CopyRecord::getId() const
{
    return _id;
}

void CegoDatabaseManager::CopyRecord::setId(int id)
{
    _id = id;
}


CegoDatabaseManager::CopyRecord& CegoDatabaseManager::CopyRecord::operator = ( const CegoDatabaseManager::CopyRecord& cr)
{
    _id = cr._id;
    _tableSet = cr._tableSet;
    _targetHost = cr._targetHost;
    _mediatorHost = cr._mediatorHost;

    _user = cr._user;
    _passwd = cr._passwd;
    _msg = cr._msg;

    return (*this);   
}

bool CegoDatabaseManager::CopyRecord::operator == ( const CegoDatabaseManager::CopyRecord& cr)
{
    if ( _tableSet == cr._tableSet && _targetHost == cr._targetHost)
	return true;
    return false;
}

/////////////////////
// DbSessionRecord //
/////////////////////

CegoDatabaseManager::DbSessionRecord::DbSessionRecord()
{
}

CegoDatabaseManager::DbSessionRecord::DbSessionRecord(CegoDistDbHandler* pHandler)
{
    _pHandler = pHandler;
}

CegoDatabaseManager::DbSessionRecord::DbSessionRecord(const Chain& hostName, const Chain& tableSet, 
				     const Chain& userName, CegoDistDbHandler* pHandler)
{
    _hostName = hostName;
    _tableSet = tableSet;
    _userName = userName;
    _pHandler = pHandler;
    _lastUsed = Datetime().asInt();
}

CegoDatabaseManager::DbSessionRecord::~DbSessionRecord()
{
}

const Chain& CegoDatabaseManager::DbSessionRecord::getHostName() const
{
    return _hostName;
}

const Chain& CegoDatabaseManager::DbSessionRecord::getTableSet() const
{
    return _tableSet;
}

const Chain& CegoDatabaseManager::DbSessionRecord::getUserName() const
{
    return _userName;
}

CegoDistDbHandler* CegoDatabaseManager::DbSessionRecord::getDbHandler() const
{
    return _pHandler;
}

bool CegoDatabaseManager::DbSessionRecord::isUsed() const
{
    return _isUsed;
}
void CegoDatabaseManager::DbSessionRecord::setUsed(bool isUsed)
{
    _isUsed = isUsed;
}

int CegoDatabaseManager::DbSessionRecord::getTSLastUsed()
{
    return _lastUsed;
}

void CegoDatabaseManager::DbSessionRecord::setTSLastUsed(int ts)
{
    _lastUsed = ts;
}

CegoDatabaseManager::DbSessionRecord& CegoDatabaseManager::DbSessionRecord::operator = ( const CegoDatabaseManager::DbSessionRecord& sr)
{

    _hostName = sr._hostName;
    _tableSet = sr._tableSet;
    _userName = sr._userName;
    _pHandler = sr._pHandler;
    _lastUsed = sr._lastUsed;    
    return (*this);
    
}

bool CegoDatabaseManager::DbSessionRecord::operator == ( const CegoDatabaseManager::DbSessionRecord& sr)
{
    if ( _pHandler == sr._pHandler )
	return true;
    return false;
}

/////////////////////////
// CegoDatabaseManager //
/////////////////////////

CegoDatabaseManager::CegoDatabaseManager(const Chain& xmlDef, const Chain& lckFileName, const Chain& logFile, CegoDbHandler::ProtocolType protType) :  CegoBufferPool(xmlDef, logFile)
{

    _protType = protType;

    _lckFileName = lckFileName;

    File lckFile(_lckFileName);
    if ( lckFile.exists() )
    {
	Chain msg = Chain("Running database instance detected at lock file ") + lckFileName;
	throw Exception(EXLOC, msg);
    }
    else
    {	
	lckFile.open(File::WRITE);	
	lckFile.writeChain("Locked");
	_nextBeat=0;
	lckFile.close();
    }

    dbmLock.init(LCKMNG_LOCKWAITDELAY, __lockStatOn);  
    for ( int i=0; i<TABMNG_MAXTABSET; i++)
    {
	_recoveryMode[i] = OFF;
    }
    _nextCopyId=1;

    _logConfigured=false;

    _modId = getModId("CegoDatabaseManager");
}

CegoDatabaseManager::~CegoDatabaseManager()
{
    File lckFile(_lckFileName);
    lckFile.remove();
}

void CegoDatabaseManager::beat()
{
    File lckFile(_lckFileName);
    lckFile.open(File::WRITE);
    lckFile.writeChain(Chain("Beat=") + Chain(_nextBeat) + Chain("\n"));	
    _nextBeat++;
    lckFile.close();    
}

void CegoDatabaseManager::PR()
{
    dbmLock.readLock(DBM_LOCKTIMEOUT);
}

void CegoDatabaseManager::PW()
{
    dbmLock.writeLock(DBM_LOCKTIMEOUT);
}

void CegoDatabaseManager::V()
{
    dbmLock.unlock();
}

bool CegoDatabaseManager::isLoggerConfigured()
{
    return _logConfigured;
}

CegoDatabaseManager::RecoveryMode CegoDatabaseManager::getRecoveryMode(int tabSetId)
{
    return _recoveryMode[tabSetId];
}

void CegoDatabaseManager::setRecoveryMode(int tabSetId, CegoDatabaseManager::RecoveryMode m)
{
    _recoveryMode[tabSetId] = m;
}

void CegoDatabaseManager::setAllRecoveryOff()
{    
    for ( int i=0; i<TABMNG_MAXTABSET; i++)
    {
	_recoveryMode[i] = OFF;
    }
}

void CegoDatabaseManager::startRecovery(const Chain& tableSet)
{
    PW();
    _recoveryList.Insert(tableSet);
    V();
}

void CegoDatabaseManager::startCopy(const Chain& tableSet, const Chain& targetHost, const Chain& mediatorHost, const Chain& user, const Chain& passwd, const Chain& msg)
{
    PW();
    _copyList.Insert( CopyRecord( tableSet, targetHost, mediatorHost, user, passwd, msg));
    V();
}

bool CegoDatabaseManager::nextRecovery(Chain& tableSet)
{
    PW();
    Chain *pS = _recoveryList.First();
    if ( pS )
    {
	tableSet = *pS;
	_recoveryList.Remove(tableSet);
	V();
	return true;
    }
    V();
    return false;
}


bool CegoDatabaseManager::nextCopy(int& id, Chain& tableSet, Chain& targetHost, Chain& mediatorHost, Chain& user, Chain& passwd)
{
    PW();
    CopyRecord *pCR = _copyList.First();
    while ( pCR )
    {
	if ( pCR->getId() == 0 )
	{
	    id = _nextCopyId++;
	    pCR->setId(id);

	    tableSet = pCR->getTableSet();
	    targetHost = pCR->getTargetHost();
	    mediatorHost = pCR->getMediatorHost();
	    user = pCR->getUser();
	    passwd = pCR->getPasswd();
	    
	    // _copyList.Remove(*pCR);
	    V();
	    return true;
	}
	pCR = _copyList.Next();
    }
    V();
    return false;
}

void CegoDatabaseManager::setCopyStatus(int id, const Chain& msg)
{
    PW();
    CopyRecord *pCR = _copyList.First();
    while ( pCR )
    {
	if ( pCR->getId() == id )
	{
	    pCR->setMsg(msg);

	    V();
	    return;
	}
	pCR = _copyList.Next();
    }
    V();
    return;

}

Element* CegoDatabaseManager::getCopyInfo()
{
    Element* pCopyInfo = new Element(XML_COPYINFO_ELEMENT);
    
    PR();
    CopyRecord *pCR = _copyList.First();
    while ( pCR )
    {
	
	Element *pN = new Element(XML_COPY_ELEMENT);

	pN->setAttribute(XML_CID_ATTR, pCR->getId());
	pN->setAttribute(XML_HOSTNAME_ATTR, pCR->getTargetHost());
	pN->setAttribute(XML_TABLESET_ATTR, pCR->getTableSet());
	pN->setAttribute(XML_STATUS_ATTR, pCR->getMsg());
	
	pCopyInfo->addContent(pN);

	pCR = _copyList.Next();
    }

    V();

    return pCopyInfo;
}


void CegoDatabaseManager::useObject(int tabSetId, const Chain& objName, CegoObject::ObjectType type, ObjectUseMode mode, long tid)
{


    if ( mode == SHARED )
    {
	// cout << "Using object " << objName << " " << type << " shared" << endl;
        PW();
        ObjectRecord *pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));
        if ( pOR == 0 )
        {
            V();
	    Chain msg = Chain("Cannot access object ") + objName;
            throw Exception(EXLOC, msg); 
        }

	if ( pOR->getTid() != 0 && pOR->getTid() == tid ) 
	{
	    pOR->incUsed();
	    V();
	}
	else
	{
	    while ( pOR && ( pOR->getMode() == EXCLUSIVE || pOR->getMode() == EXCLUSIVE_WRITE ) )
	    {
		V();
		PW();
		pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));
	    }
	    if ( pOR == 0 )
	    {
		V();
		Chain msg = Chain("Cannot access object ") + objName;
		throw Exception(EXLOC, msg); 
	    }
	    
	    
	    pOR->incUsed();
	    V();
	}
    }
    else if ( mode == EXCLUSIVE )
    {
	// cout << "Using object " << objName << " " << type << " exclusive" << endl;
        PW();
        ObjectRecord *pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));

        if ( pOR == 0 )
        {
            V();
            Chain msg = Chain("Cannot access object ") + objName;
            throw Exception(EXLOC, msg);
        }
	
	pOR->setMode(mode);
	pOR->setTid(tid);
	    
	while ( pOR && pOR->numUsed() != 0 )
	{
	    V();
	    Sleeper ns;
	    ns.nanoSleep(10000);
	    PW();
	    
	    pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));            
	}

        if ( pOR == 0 )
        {
            V();
            Chain msg = Chain("Cannot access object ") + objName;

            throw Exception(EXLOC, msg);
        }
        pOR->incUsed();
        V();
    }
    else if ( mode == EXCLUSIVE_WRITE )
    {
	// cout << "Using object " << objName << " " << type << " exclusive_write " << endl;
        PW();
        ObjectRecord *pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));

        if ( pOR == 0 )
        {
            V();
            Chain msg = Chain("Cannot access object ") + objName;
            throw Exception(EXLOC, msg);
        }
	
	// cout << "NumUsed = " << pOR->numUsed() << endl;

	// if ( pOR->getMode() == SHARED  )

	if ( pOR->getMode() == SHARED && pOR->numUsed() == 0 )
	{
	    pOR->setMode(mode);
	    pOR->setTid(tid);
	}
	else
	{	    
	    // while ( pOR && pOR->getMode() != SHARED )
	    while ( pOR && pOR->numUsed() != 0 )
	    {
		V();
		Sleeper ns;
		ns.nanoSleep(10000);
		PW();

		pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));            
	    }

	    if ( pOR == 0 )
	    {
		V();
		Chain msg = Chain("Cannot access object ") + objName;
		throw Exception(EXLOC, msg);
	    }

	    pOR->setMode(mode);

	}

        if ( pOR == 0 )
        {
            V();
	    Chain msg = Chain("Cannot access object ") + objName;
            throw Exception(EXLOC, msg);
        }
        pOR->incUsed();
        V();
    }
    return;
}

void CegoDatabaseManager::unuseObject(int tabSetId, const Chain& objName, CegoObject::ObjectType type, ObjectUseMode mode)
{
    // cout << "Unusing object " << objName << " " << type  << endl;
    PW();
    ObjectRecord *pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));

    if ( pOR == 0 )
    {
        V();
	Chain msg = Chain("Cannot access object ") + objName;
	throw Exception(EXLOC, msg);
	return;
    }
    
    pOR->decUsed();

    if ( mode != SHARED )
    {
        pOR->setMode(SHARED);
	pOR->setTid(0);
    }
    V();    
}

void CegoDatabaseManager::addObject(int tabSetId, const Chain& objName, CegoObject::ObjectType type)
{
    PW();
    _objList.Insert(ObjectRecord(tabSetId, objName, type));
    V();
}

void CegoDatabaseManager::removeObject(int tabSetId, const Chain& objName, CegoObject::ObjectType type)
{
    PW();
    _objList.Remove(ObjectRecord(tabSetId, objName, type));
    V();
}

bool CegoDatabaseManager::objectExists(int tabSetId, const Chain& objName, CegoObject::ObjectType type)
{

    PR();
    
    ObjectRecord *pTR = _objList.Find(ObjectRecord(tabSetId, objName, type));

    V();
    
    if ( pTR ) 
    {	
	return true;
    }
    else
    {
	return false;
    }    

}

void CegoDatabaseManager::setThreadInfo(int numDbThread, int numAdmThread, int numLogThread)
{
    _numDbThread = numDbThread;
    _numAdmThread = numAdmThread;
    _numLogThread = numLogThread;
    _activeDbThread = 0;
    _activeAdmThread = 0;
    _activeLogThread = 0;
}
    
void CegoDatabaseManager::getThreadInfo(int& numDbThread, int& numAdmThread, int& numLogThread,
		   int& activeDbThread, int& activeAdmThread, int& activeLogThread)
{
    numDbThread = _numDbThread;
    numAdmThread = _numAdmThread;
    numLogThread = _numLogThread;
    activeDbThread = _activeDbThread;
    activeAdmThread = _activeAdmThread;
    activeLogThread = _activeLogThread;
}

void CegoDatabaseManager::increaseActiveAdmThread()
{
    _activeAdmThread++;
}

void CegoDatabaseManager::decreaseActiveAdmThread()
{
    _activeAdmThread--;
}

void CegoDatabaseManager::increaseActiveDbThread()
{
    _activeDbThread++;
}

void CegoDatabaseManager::decreaseActiveDbThread()
{
    _activeDbThread--;
}

void CegoDatabaseManager::increaseActiveLogThread()
{
    _activeLogThread++;
}

void CegoDatabaseManager::decreaseActiveLogThread()
{
    _activeLogThread--;
}

Chain CegoDatabaseManager::getSysFileName(const Chain& tableSet)
{
    Chain tsRoot = CegoXMLSpace::getTSRoot(tableSet);
    Chain dbName = CegoXMLSpace::getDbName();
    return tsRoot + Chain("/") + dbName + Chain("_") + Chain(tableSet) + Chain(SYSSUFFIX);
}


Chain CegoDatabaseManager::getTmpFileName(const Chain& tableSet)
{
    Chain tsRoot = CegoXMLSpace::getTSRoot(tableSet);
    Chain dbName = CegoXMLSpace::getDbName();
    return tsRoot + Chain("/") + dbName + Chain("_") + Chain(tableSet) + Chain(TEMPSUFFIX);
}

void CegoDatabaseManager::getDBMLockStat(Chain& lockName, long& lockCount, long &numRdLock, long &numWrLock, long &sumRdDelay, long &sumWrDelay)
{
    lockName = dbmLock.getId();
    lockCount= dbmLock.numLockTry();

    numRdLock = dbmLock.numReadLock();
    numWrLock = dbmLock.numWriteLock();
    sumRdDelay = 0;
    sumWrDelay = 0;

    if ( dbmLock.numReadLock() > 0 )
	sumRdDelay = dbmLock.sumReadDelay() / LCKMNG_DELRES;
    if ( dbmLock.numWriteLock() > 0 )
	sumWrDelay = dbmLock.sumWriteDelay() / LCKMNG_DELRES;

}


CegoDistDbHandler* CegoDatabaseManager::allocateSession(const Chain& hostName, const Chain& tableSet,
							const Chain& userName, const Chain& password)
{

    PW();

    
    try
    {
    
	DbSessionRecord *pSR = _dbSessionList.First();
	while ( pSR )
	{
	    if ( pSR->getHostName() == hostName 
		 && pSR->getTableSet() == tableSet 
		 && pSR->getUserName() == userName 
		 && pSR->isUsed() == false )
	    {
		pSR->setUsed(true);
		Datetime dt;
		pSR->setTSLastUsed( dt.asInt() );
		
		V();
		
		return pSR->getDbHandler();
	    }
	    pSR = _dbSessionList.Next();
	}
	
	// create new session

	CegoDistDbHandler* pSession = createSession(hostName, tableSet, userName, password);
	_dbSessionList.Insert( DbSessionRecord(hostName, tableSet, userName, pSession));

	V();

	return pSession;

    }
    catch ( Exception e )
    {
	V();
	throw Exception(EXLOC, "Cannot allocate db session", e);
    }
}

void CegoDatabaseManager::releaseSession(CegoDistDbHandler* pHandler)
{

    PW();
    
    DbSessionRecord *pSR = _dbSessionList.First();
    while ( pSR )
    {
	if ( pSR->getDbHandler() == pHandler  )
	{
	    pSR->setUsed(false);
	    
	    V();
	    
	    return;
	}
	pSR = _dbSessionList.Next();
    }
    
    V();
    
    Chain msg = Chain("Cannot release session for unknown db handle"); 
    throw Exception(EXLOC, msg);

}

void CegoDatabaseManager::cleanSession(int lifetime)
{

    PW();
    
    Datetime dt;
    
    DbSessionRecord *pSR = _dbSessionList.First();
    while ( pSR )
    {
	if ( pSR->getTSLastUsed() <  ( dt.asInt() - lifetime )  )
	{
	    if ( pSR->isUsed() == false )
	    {
		closeSession(pSR->getDbHandler());
		_dbSessionList.Remove( DbSessionRecord(pSR->getDbHandler()));
		pSR = _dbSessionList.First();
	    }
	}
	pSR = _dbSessionList.Next();
    }
    
    V();
    
}


Element* CegoDatabaseManager::getSessionInfo(int lifetime)
{
    Element* pSessionInfo = new Element(XML_DBSESSIONINFO_ELEMENT);
    
    DbSessionRecord *pSR = _dbSessionList.First();
    while ( pSR )
    {
	
	Element *pN = new Element(XML_DBSESSION_ELEMENT);
	pN->setAttribute(XML_HOSTNAME_ATTR, pSR->getHostName());
	pN->setAttribute(XML_TABLESET_ATTR, pSR->getTableSet());
	pN->setAttribute(XML_USER_ATTR, pSR->getUserName());
	if ( pSR->isUsed() )
	    pN->setAttribute(XML_ISUSED_ATTR, XML_TRUE_VALUE);
	else
	    pN->setAttribute(XML_ISUSED_ATTR, XML_FALSE_VALUE);

	Datetime dt;
	pN->setAttribute(XML_TTL_ATTR, Chain(pSR->getTSLastUsed() + lifetime - dt.asInt()));	

	pSessionInfo->addContent(pN);

	pSR = _dbSessionList.Next();
    }


    return pSessionInfo;
}


CegoDistDbHandler* CegoDatabaseManager::createSession(const Chain& hostName, const Chain& tableSet,
						  const Chain& userName, const Chain& password)
{
    int portNo;
    getDataPort(portNo);
    
    Net n( NETMNG_MSG_BUFLEN, NETMNG_SIZEBUFLEN );
    NetHandler* pN = 0;
    CegoDistDbHandler *pSH = 0;

    try     
    {

#ifdef DEBUG
	log(_modId, Logger::DEBUG, Chain("Connecting to ") + Chain(hostName) + Chain(" on port ") + Chain(portNo));
#endif

	pN = n.connect(hostName, portNo);
	pSH = new CegoDistDbHandler(pN, _protType, this);
	
#ifdef DEBUG
	log(_modId, Logger::DEBUG, Chain("Using user ") + userName + Chain("/") + password); 
#endif
	pSH->requestSession(tableSet, userName, password, false);

    }
    catch ( Exception e ) 
    {
	
	if ( pSH )
	{
	    delete pSH;
	}
	
	if ( pN )
	{
	    delete pN;
	}
	
	Chain msg = Chain("Cannot create session to ") + hostName;
	throw Exception(EXLOC, msg);
    }

    return pSH;
}

void CegoDatabaseManager::closeSession(CegoDistDbHandler* pSH)
{
#ifdef DEBUG
    log(_modId, Logger::DEBUG, Chain("Closing session ..."));     
#endif

    pSH->closeSession();

    NetHandler *pN = pSH->getNetHandler();

    delete pSH;
    delete pN;

}

void CegoDatabaseManager::configureLogger(Logger::LogLevel level)
{
    for ( int i=1; i< getMapSize() ; i++)
    {
	logModule(i, getModName(i), level);
    }
    _logConfigured=true;
}
    
void CegoDatabaseManager::configureLogger()
{
    ListT<Chain> modList;
    _logConfigured =  getModuleList(modList);

    Chain *pMod = modList.First();
    while ( pMod )
    {
	if ( (Chain)pMod->toUpper() == Chain("ALL"))
	{
	    Logger::LogLevel level = getLogLevel(*pMod);
	    for ( int i=1; i< getMapSize() ; i++)
	    {
		logModule(i, getModName(i), level);
	    }
	}
	else
	{
	    unsigned long modId = getModId(*pMod);
	    logModule(modId, *pMod, getLogLevel(*pMod));
	}
	pMod = modList.Next();
    }
}

bool CegoDatabaseManager::verifyAccess(const int tabSetId, const Chain& objName, CegoObject::ObjectType type, CegoXMLSpace::AccessMode mode, const Chain& user)
{

    SetT<Chain> roleSet;

    getRoleSet(user, roleSet);
    Chain tableSet = getTabSetName(tabSetId);

    Chain *pRole = roleSet.First();
    while ( pRole ) 
    {
	
	Chain objPattern = objName;

	if ( matchRole( *pRole, tableSet, objPattern, mode ) )
	    return true;
	pRole = roleSet.Next();
    }
    return false;
}

void CegoDatabaseManager::initLogFiles(const Chain& tableSet, bool overwrite)
{

    ListT<Chain> lfList;
    ListT<int> sizeList;
    ListT<Chain> statusList;
    
    int tabSetId = getTabSetId(tableSet);
    getLogFileInfo(tableSet, lfList, sizeList, statusList);

    Chain *pLog = lfList.First();
    int *pSize = sizeList.First();

    bool isFirst = true;
    while ( pLog ) 
    {
	
	if ( isFirst )
	    setLogFileStatus(tableSet, *pLog, XML_ACTIVE_VALUE);		
	else
	    setLogFileStatus(tableSet, *pLog,  XML_FREE_VALUE);	
	
	isFirst=false;

	log(_modId, Logger::NOTICE, Chain("Initializing logfile ") + *pLog + Chain(" ..."));
	
	if ( overwrite == false )
	{
	    File checkLog(*pLog);
	    if ( checkLog.exists() ) 
	    {
		Chain msg = Chain("Cannot initialize logfile <") + *pLog + Chain(">, file already exists");
		throw Exception(EXLOC, msg);
	    }
	}

	setLogFile(tabSetId, *pLog, false);
	initLog(tabSetId, *pSize);
	
	pLog = lfList.Next();
	pSize = sizeList.Next();
	
    }
}

void CegoDatabaseManager::releaseLogFiles(const Chain& tableSet, bool waitForArch)
{

    ListT<Chain> lfList;
    ListT<int> sizeList;
    ListT<Chain> statusList;
    
    int tabSetId = getTabSetId(tableSet);
    getLogFileInfo(tableSet, lfList, sizeList, statusList);

    Chain *pLog = lfList.First();
    Chain *pStatus = statusList.First();

    while ( pLog && pStatus ) 
    {	

	if ( *pStatus == Chain(XML_ACTIVE_VALUE) )
	{
	    setLogFile(tabSetId, *pLog, true);
	    long minlsn = getMinLSN(tabSetId);
	    if ( minlsn > 0 )
	    {
		log(_modId, Logger::NOTICE, Chain("Releasing logfile ") + *pLog + Chain(" LSN=") + Chain(minlsn));
		setLogFileStatus(tableSet, *pLog, XML_OCCUPIED_VALUE);
	    }
	}
	pStatus = statusList.Next();
	pLog = lfList.Next();	
    }


    if ( waitForArch )
    {

	bool notArchived = true;

	while ( notArchived )
	{

	    log(_modId, Logger::NOTICE, Chain("Waiting for archive ... "));	    

	    ListT<Chain> lfList;
	    ListT<int> sizeList;
	    ListT<Chain> statusList;
	
	    getLogFileInfo(tableSet, lfList, sizeList, statusList);
	    
	    notArchived = false;

	    Chain *pStatus = statusList.First();
	    while ( pStatus ) 
	    {	
		if ( *pStatus != Chain(XML_FREE_VALUE) )
		    notArchived = true;
		pStatus = statusList.Next();
	    }


	    lfList.Empty();
	    sizeList.Empty();
	    statusList.Empty();

	    Sleeper s;
	    s.secSleep(LOGMNG_RECOVERY_DELAY);

	}
    }

}





