// Change History:

#include "ddp/ddp_session.h"
#include "ddp/ddp_process.h"

// **************************************************************************
// DDP_Session
// **************************************************************************

DDP_Session::DDP_Session() : App(NULL), VectorCount(0),
    PeriodCount(0), hPush(NULL), NextID(0)
{
}
//--------------------------------------------------------------------------

DDP_Session::~DDP_Session()
{
    if(hPush)
        {
        SetEvent(hPush);
        CloseHandle(hPush);
        }
        
    hPush = NULL;
    Clear();
}
//--------------------------------------------------------------------------

bool DDP_Session::Load(const DRS_Application * appl)
{
    App = appl;
    if(!App)
        {
        PushError("Failed to load session. Application is NULL");
        return false;
        }

    // load application data
    sID = (char *)App->ID();
    sDescription = (char *)App->Description();

    sINIFile = (char *)App->INIFile();

    VectorCount = App->VectorCount();

    // load custom paths
    std::string fn;

    // load database info

    uint i, cnt = App->DatabaseCount();
    Databases.resize(cnt);

    byte buf[500];
    std::string temp;
    for(i = 0; i < cnt; ++i)
        {
        const DRS_Database * database = &App->Database(i);
        Databases[i].ID = (char)database->Designator();
        if(database->FieldCount())
            {
            Databases[i].IsScalar = true;
            PORT_snprintf_1( buf, sizeof(buf), "%sF", Databases[i].ID.c_str());
            }
        else
            {
            PeriodCount = max(PeriodCount, database->CoreDatabase()->getPeriods());
            PORT_snprintf_1( buf, sizeof(buf), "%sV", Databases[i].ID.c_str());
            Databases[i].IsScalar = false;
            }

        temp = (char *)buf;
        DRS_Application::DataFilePathGet( PORT_c_str(temp), buf, sizeof(buf), PORT_c_str(sINIFile));
        DRS_Application::DataFileAssert(buf);
        Databases[i].ConfigFile = (char *)buf;
        }

    // Load tree and layout
    if(!AppVectorTree.LoadTree(appl))
        {
        PushError("Failed to load session for application: %s.\nVector Database definition cannot be loaded.",
            appl->Description());
        return false;
        }
    if(!AppLayout.Load(App))
        {
        PushError("Failed to load session for application: %s.\nScalar Database definition cannot be loaded.",
        App->Description());
        return false;
        }

return true;
}
//--------------------------------------------------------------------------
std::vector<DDP_Session *> DDP_Session::SessionPool;
//--------------------------------------------------------------------------

DDP_Session * DDP_Session::PushSession()
{
    DDP_Session * session = new DDP_Session;
    SessionPool.push_back(session);

return session;
}
//--------------------------------------------------------------------------

DDP_Session * DDP_Session::GetSession(DRS_Application * a)
{
    uint i, cnt = SessionPool.size();
    for(i = 0; i < cnt; ++i)
        if(SessionPool[i]->App == a)
            return SessionPool[i];
            
return NULL;
}
//--------------------------------------------------------------------------

bool DDP_Session::PopSession(DRS_Application * a)
{
    DDP_Session * session = GetSession(a);
    if(!session)
        return false;

    // pop session
    uint i, cnt = SessionPool.size();
    for(i = 0; i < cnt; ++i)
        if(session == SessionPool[i])
            {
            SessionPool[i] = NULL;
            SessionPool.erase(SessionPool.begin() + i);
            break;
            }
    // delete session
    delete session;

return true;
}
//--------------------------------------------------------------------------

bool DDP_Session::FindSession(DDP_Session * session)
{
    uint i, cnt = SessionPool.size();
    for(i = 0; i < cnt; ++i)
        if(session == SessionPool[i])
            return true;

return false;            
}
//--------------------------------------------------------------------------

bool DDP_Session::DumpRecList(const AT_Record_List *pRecordList, const byte * file, bool keepOrder)
{
  if(!pRecordList)
    {
    PushError("Failed to dump record list.");
    return false;
    }

  AT_Write_Seq_File * rec_list(ARL_CreateWriteSeqFile(file, ATFT_Raw));
  AT_Record_List::Iter * it = ARL_CreateRecordListIter(pRecordList->begin());
  std::vector<bool> RecMap;

  if(!keepOrder)
    RecMap.resize(pRecordList->dbRecords(), false);

  const DDP_Params * params = GetParams();
  uint i, cnt = params ? params->GetInt((const byte *)"RecordCount") : pRecordList->count();
  uint32 rec;
  for(i = 0; (i < cnt) && it && *(*it) != MAXUINT; ++(*it), ++i)
    {
    rec = *(*it);
    if(keepOrder)
        rec_list->write(&rec, sizeof(uint32));
    else
        RecMap[rec] = true;
    }

  if(!keepOrder)
    {
      rec = 0;
      uint rmsz = RecMap.size();
      while(cnt && rec < rmsz)
        {
        if(RecMap[rec])
            {
            rec_list->write(&rec, sizeof(uint32));
            --cnt;
            }
        ++rec;
        }
    }

  ARL_DeleteRecordListIter(it);
  ARL_DeleteFile(rec_list);

  return true;
}
//--------------------------------------------------------------------------

bool DDP_Session::DumpReportParams(AT_Record_List *pRecordList, const DDP_ExportDef** Fields,
    const DDP_ExportDef** Vectors)
{
    DDP_Report * report = new DDP_Report(this);
    if(!pRecordList)
        {
        delete report;
        PushError("Failed to execute report. Invalid record list");
        return false;
        }
    report->SetRecList(pRecordList);
    bool res = report->DumpParams(Fields, Vectors);
    if(!res)
        {
        delete report;
        PushError("Failed to execute report. Invalid report options");
        return false;
        }
    PushProcess(report);

return true;
}
//--------------------------------------------------------------------------

bool DDP_Session::dump_export(DDP_Export * exp, const void** Fields,
    const byte * FieldTypes, const byte * FileName, const byte * ExpType)
{
    if(!exp)
        {
        PushError("Failed to execute export. Export process is not valid");
        return false;
        }

    exp->SetExportFile(FileName);
    if(!exp->SetExportType(ExpType))
        {
        PushError("Failed to execute browse export. Unknown export type");
        return false;
        }

    uint cnt = 0;
    while(Fields && Fields[cnt])
        {
        if(FieldTypes && FieldTypes[cnt] == FIELD_TYPE_SCALAR)
            {
            const DDP_Field * field = (const DDP_Field *)Fields[cnt];
            exp->AddScalar(field);
            exp->AddVector(NULL);
            }
        else if(FieldTypes)
            {
            exp->AddScalar(NULL);
            exp->AddVector((const DDP_VectorDef *)Fields[cnt]);
            }
        ++cnt;
        }

    bool res = exp->Dump();
    if(!res)
        {
        PushError("Failed to execute export. Invalid export options");
        return false;
        }

    PushProcess(exp);
    
return true;
}
//--------------------------------------------------------------------------

bool DDP_Session::DumpBrowseParams(const AT_Index *pIndex,
    const void** Fields, const byte * FieldTypes, const byte * FileName,
    const byte * ExpType)
{
    DDP_BrowseExport * piexport = new DDP_BrowseExport(this);
    if(!pIndex)
        {
        delete piexport;
        PushError("Failed to execute browse export. Invalid index");
        return false;
        }
    piexport->SetIndex(pIndex);

    // add frequency field
    DDP_Field occurs((const byte *)"#FREQUENCY#");
    occurs.SetType(DATAFIELDTYPE_INTEGER);
    occurs.SetSize(10);
    piexport->AddScalar(&occurs);
    piexport->AddVector(NULL);
    if(!dump_export(piexport, Fields, FieldTypes, FileName, ExpType))
        {
        delete piexport;
        return false;
        }

return true;
}
//--------------------------------------------------------------------------

bool DDP_Session::DumpExportParams(AT_Record_List *pRecordList,
    const void** Fields, const byte * FieldTypes, const DDP_Field** SplitFields,
    const byte * FileName, const byte * ExpType, bool AutoSave)
{
    DDP_DataExport * pexport = new DDP_DataExport(this);

    // set record list
    pexport->SetRecList(pRecordList);
    
    pexport->SetMultiFile(SplitFields);
    pexport->SetAutoSave(AutoSave);
    if(!dump_export(pexport, Fields, FieldTypes, FileName, ExpType))
        {
        delete pexport;
        return false;
        }
        
return true;
}
//--------------------------------------------------------------------------

DDP_Session::Job * DDP_Session::push_job()
{
    // new job
    Jobs.push_back(Job());
return &Jobs.back();
}
//--------------------------------------------------------------------------

uint DDP_Session::push_params(DDP_Params * params)
{
    if(!params)
        return 0;

    // wait for start to complete
    if(hPush && WaitForSingleObject(hPush, INFINITE) == WAIT_FAILED)
        throw AT_System_Err;

    Job * job = push_job();
    // new params
    job->Params = params;

return NextID++;
}
//--------------------------------------------------------------------------

const DDP_Params * DDP_Session::GetParams() const
{
    uint cnt = GetJobCount();
    if(!cnt)
        // create new params
        return NULL;

    const Job * job = &Jobs.back();
    if(job->Process)
        // process already created. Too late for params!!!
        return NULL;

return job->Params;
}
//--------------------------------------------------------------------------

uint DDP_Session::PushProcess(DDP_Process * process)
{
    // wait for start to complete
    if(hPush && WaitForSingleObject(hPush, INFINITE) == WAIT_FAILED)
        throw AT_System_Err;

    uint cnt = GetJobCount();
    uint id = NextID;
    Job * job = NULL;

    if(!cnt || Jobs[cnt - 1].Process)
        {
        job = push_job();
        ++NextID;
        }

    job = job ? job : &Jobs.back();
        
    job->Process = process;
    process->RaiseEvent();
        
return id;
}
//--------------------------------------------------------------------------

DDP_Process * DDP_Session::PopProcess()
{
    // wait for prior start to complete
    if(hPush && WaitForSingleObject(hPush, INFINITE) == WAIT_FAILED)
        throw AT_System_Err;
        
    DDP_Process * process = FrontProcess();
	std::vector<Job>::iterator b;
    if(process)
        {
        b = Jobs.begin();
        if(b->Params)
            delete b->Params;
        Jobs.erase(b);
        }
        
return process;
}
//--------------------------------------------------------------------------

const DDP_Process * DDP_Session::WaitForProcess(const DDP_Process * process) const
{
    uint i, cnt = GetJobCount();
    const Job * prev_job = NULL;

    for(i = 0; i < cnt; ++i)
        {
        const Job * job = get_job(i);
        if(job->Process == process)
            return prev_job ? prev_job->Process : NULL;
        prev_job = job;
        }

return prev_job ? prev_job->Process : NULL;        
}
//--------------------------------------------------------------------------

uint64 DDP_Session::GetFileSize(const byte * fname)
{
    uint64 res;
    try
    {
    res = AT_File::GetFileSize(fname);
    }
    catch(...)
    {
        return 0;
    }
return res;
}
//--------------------------------------------------------------------------

bool DDP_Session::LogProcessTime(const DDP_Process * process, uint open, const byte * fmt, ... )
{
    if(!process)
        return false;

    const byte * fname = process->GetProcessLogFile(open != LOG_OPEN);
    if(!fname)
        return false;

    char buf[DDP_PrintFile::LINE_SIZE];
    char * line = buf;

    uint sz = DDP_Process::TimeStamp(PORT_byte(line), sizeof(buf));
    line += sz;
    *line = ':';
    ++sz;
    *(line + 1) = ' ';
    ++sz;
    
    if(fmt)
        {
        va_list arg;
        va_start(arg, fmt);
        sz += vsnprintf((buf + sz), sizeof(buf) - sz, PORT_cchar(fmt), arg);
        va_end(arg);
        }

    if((open == LOG_WRITE || open == LOG_CLOSE) &&
        !process->GetProcessLogFileSize(open != LOG_OPEN))
        return true;
        
    FILE * log = std::fopen(PORT_cchar(fname), "a");
    if(!log)
        return false;
        
    if(open == LOG_OPEN)
        fprintf(log, "\r\n\r\n[Process Time]\r\n");
    fprintf(log, "%s\r\n", buf);
    fclose(log);
      
return true;
}
//----------------------------------------------------------------------------

bool DDP_Session::Start(const DDP_Process * process)
{
    // wait for prior start to complete
    if(hPush && WaitForSingleObject(hPush, INFINITE) == WAIT_FAILED)
        throw AT_System_Err;

    hPush = CreateEvent(NULL, TRUE, TRUE, NULL);
    // raise push event
    ResetEvent(hPush);
    const DDP_Process * wait_process = WaitForProcess(process);
    // release push event
    SetEvent(hPush);
    CloseHandle(hPush);
    hPush = NULL;

    if(wait_process)
        {
        LogProcessTime(process, DDP_Session::LOG_OPEN, PORT_cbyte("Waiting in Process Queue..."));
        HANDLE h = wait_process->GetEvent();
        // waite for process
        if(h)
            {
            if(WaitForSingleObject(h, INFINITE) == WAIT_FAILED)
                throw AT_System_Err;
            }
        }
        
    if(!GetJobCount())
        return false;
        
    DDP_Process * front = FrontProcess();
    if(front)
        {
        front->SetActive(true);
        LogProcessTime(front,
            (!wait_process ? DDP_Session::LOG_OPEN : DDP_Session::LOG_WRITE), PORT_cbyte("Start Process") );
        }
return true;
}

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

bool DDP_Session::Stop()
{
    // remove process from cache
    DDP_Process * process = PopProcess();

    if(process)
        {
        process->ReloadEnv();

        LogProcessTime(process, DDP_Session::LOG_CLOSE, PORT_cbyte("Stop Process") );

        bool cleanupTemp = process->IsActive();
        process->SetActive(false);
        process->Cleanup(cleanupTemp);
        delete process;
        }

return true;
}
//--------------------------------------------------------------------------

bool DDP_Session::Execute()
{
    DDP_Process * process = BackProcess();
    if(!process)
        return true;
return process->Execute();
}
//--------------------------------------------------------------------------

void DDP_Session::Clear()
{
    std::vector<DDP_Process *> trash;

    // empty process cache
    while(GetJobCount())
        {
        DDP_Process * process = PopProcess();
        if(process)
            trash.push_back(process);
        }

    // delete all processes
    uint i, cnt = trash.size();
    for(i = 0; i < cnt; ++i)
        delete trash[i];
}
//--------------------------------------------------------------------------

uint DDP_Session::GetCachedProcessNames(byte * buf, uint sz) const
{
    if(!buf || !sz)
        return 0;
    uint n = GetJobCount();
    if(!n || n == 1)
        return 0;
    uint name_sz = sz / n;
    if(!name_sz)
        return 0;
    byte *p = buf;
    memset(p, 0, sz);
    for(std::vector<Job>::const_iterator it = Jobs.begin();
        it != Jobs.end(); it++)
        {
        const Job j = (*it);
        const DDP_Process * process = j.Process;
        if(process)
            {
            uint s = PORT_snprintf_2(p, sz, "%s: %s", process->GetProcess(),
                process->GetProcessTime());
            ++s;
            if(s > name_sz)
                {
                p += (name_sz - 1);
                *p = 0;
                ++p;
                }
            else
                p += s;
            }
        }
return n;
}
//--------------------------------------------------------------------------

bool DDP_Session::push_err(const byte * err)
{
    if(!err)
        return false;

    Errors.push_back((char *)err);
    
return true;
}
//--------------------------------------------------------------------------

bool DDP_Session::PushError(AT_Error & err)
{
    for (AT_Error *e = &err; e; e = e->getCause())
        if(!push_err(e->getObjectName()))
            return false;
return true;            
}
//--------------------------------------------------------------------------

bool DDP_Session::PushError(const char * fmt, ... )
{
    if(!fmt || !*fmt)
        return false;
    char buf[ERROR_SIZE];
    va_list arg;
    va_start(arg, fmt);
    vsnprintf(buf, sizeof(buf), fmt, arg);
    va_end(arg);
    if(!push_err(PORT_cbyte(buf) ))
        return false;
        
return true;
}
//--------------------------------------------------------------------------

uint DDP_Session::PopError(byte * buf, uint sz)
{
    if(!buf || !sz)
        return 0;

    if(!GetErrorCount())
        return 0;

    uint len = Errors.back().length();
    PORT_strncpy(buf, Errors.back().c_str(), sz);
    Errors.pop_back();
    
return len;
}

bool DDP_Session::SetParam(const byte * key, const byte * val)
{
    DDP_Params * params = (DDP_Params *)GetParams();
    if(!params)
        return false;

    params->SetParam(key, val);

return true;
}

//--------------------------------------------------------------------------
// eof

/****************************************************************************
 NOTES
 ****************************************************************************/