/**
 * Tencent is pleased to support the open source community by making Tars available.
 *
 * Copyright (C) 2016THL A29 Limited, a Tencent company. All rights reserved.
 *
 * Licensed under the BSD 3-Clause License (the "License"); you may not use this file except 
 * in compliance with the License. You may obtain a copy of the License at
 *
 * https://opensource.org/licenses/BSD-3-Clause
 *
 * Unless required by applicable law or agreed to in writing, software distributed 
 * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR 
 * CONDITIONS OF ANY KIND, either express or implied. See the License for the 
 * specific language governing permissions and limitations under the License.
 */

#include "QueryDbThread.h"
#include "QueryServer.h"
#include "DbProxy.h"

/////////////////////////////////////////////////////////////////////////
QueryDbThread::QueryDbThread()
: _terminate(false)
{
    TLOGDEBUG("QueryDbThread init ok" << endl);
}

QueryDbThread::~QueryDbThread()
{
    terminate();

    TLOGDEBUG("QueryDbThread terminate." << endl);
}

void QueryDbThread::start(int iThreadNum)
{
    for(int i = 0; i < iThreadNum; ++i)
    {
        HandleThreadRunner *r = new HandleThreadRunner(this);

        r->start();

        _runners.push_back(r);
    }
}

void QueryDbThread::terminate()
{
    _terminate = true;

    for (uint32_t i = 0; i < _runners.size(); ++i)
    {
        if (_runners[i]->isAlive())
        {
            _runners[i]->terminate();

            _queue.notifyT();
        }
    }

    for (uint32_t i = 0; i < _runners.size(); ++i)
    {
        if(_runners[i]->isAlive())
        {
            _runners[i]->getThreadControl().join();
        }
    }

    _queue.clear();

    for (uint32_t i = 0; i < _runners.size(); ++i)
    {
        if(_runners[i])
        {
            delete _runners[i];
            _runners[i] = NULL;
        }
    }
}

bool QueryDbThread::pop(QueryItem* &pItem)
{
    return _queue.pop_front(pItem, 2000);
}

void QueryDbThread::put(QueryItem* pItem)
{
    if(!_terminate && pItem)
    {
        _queue.push_back(pItem);
    }
}

/////////////////////////////////////////////////////////////////////////
HandleThreadRunner::HandleThreadRunner(QueryDbThread* proc)
: _terminate(false)
, _proc(proc)
{
}

void HandleThreadRunner::terminate()
{
    _terminate = true;
}

void HandleThreadRunner::run()
{
    string sRes("");
    DbProxy    _dbproxy;
    int64_t tStart    = 0;
    int64_t tEnd    = 0;

    while (!_terminate)
    {
        QueryItem* pQueryItem = NULL;

        if(!_terminate && _proc->pop(pQueryItem))
        {
            try
            {
                tStart    = TNOWMS;

                _dbproxy.queryData(pQueryItem->mQuery, sRes, pQueryItem->bFlag);

                tEnd    = TNOWMS;

                sRes += "endline\n";

                pQueryItem->current->sendResponse(sRes.c_str(), sRes.length());

                FDLOG("inout") << "HandleThreadRunner::run sUid:" << pQueryItem->sUid << "queryData  timecost(ms):" << (tEnd - tStart) << endl;
            }
            catch(exception& ex)
            {
                TLOGERROR("HandleThreadRunner::run exception:" << ex.what() << endl);

                string sResult = "Ret:-1\n" +  string(ex.what()) + "\nendline\n";
                pQueryItem->current->sendResponse(sResult.c_str(), sResult.length());

                FDLOG("inout") << "HandleThreadRunner::run sUid:" << pQueryItem->sUid << "exception:" << ex.what() << endl;
            }
            catch(...)
            {
                TLOGERROR("HandleThreadRunner::run exception." << endl);

                string sResult = "Ret:-1\nunknown exception\nendline\n";
                pQueryItem->current->sendResponse(sResult.c_str(), sResult.length());

                FDLOG("inout") << "HandleThreadRunner::run sUid:" << pQueryItem->sUid << "unknown exception." << endl;
            }

            sRes = "";

            delete pQueryItem;
            pQueryItem = NULL;
        }
    }
}