/**
 * 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 "util/tc_md5.h"
#include "BatchPatchThread.h"
#include "ServerFactory.h"
#include "CommandPatch.h"
#include "NodeRollLogger.h"
#include "util.h"

using namespace tars;

BatchPatch::BatchPatch()
{
 
}

BatchPatch::~BatchPatch()
{
 terminate();
}

void BatchPatch::start(int iNum)
{
 for (int i = 0; i < iNum; i++)
 {
 BatchPatchThread * t = new BatchPatchThread(this);
 t->setPath(_downloadPath);
 t->start();

 _runners.push_back(t);
 }
}

void BatchPatch::terminate()
{
 for (size_t i = 0; i < _runners.size(); ++i)
 {
 if(_runners[i]->isAlive())
 {
 _runners[i]->terminate();
 }
 }

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

}

void BatchPatch::push_back(const tars::PatchRequest & request, ServerObjectPtr serverPtr)
{
 {
 TC_ThreadLock::Lock LockQueue(_queueMutex);

 if (_patchIng.count(request.appname + request.servername) == 1)
 {
 std::string sException = "reduplicate patch request:" + request.version + "," + request.user;

 NODE_LOG("patchPro")->debug() << FILE_FUN<< request.appname + "." + request.servername << "|" << sException << endl;
 throw TC_Exception(sException);
 }

 if (!serverPtr)
 {
 NODE_LOG("patchPro")->error() << FILE_FUN<< "server null " <<endl;
 throw TC_Exception("sever null");
 }

 _patchQueue.push_back(make_pair(request,serverPtr));

 _patchIng.insert(request.appname + request.servername);
 }

}

bool BatchPatch::pop_front(pair<tars::PatchRequest,ServerObjectPtr>& item)
{
 TC_ThreadLock::Lock LockQueue(_queueMutex);

 bool bRet=false;

 bRet=_patchQueue.pop_front(item);

 if(bRet)
 {
 _patchIng.erase(item.first.appname+item.first.servername);
 }

 return bRet;
}

void BatchPatch::timedWait(int millsecond)
{
 TC_ThreadLock::Lock lock(_queueMutex);
 
 _queueMutex.timedWait(millsecond);
}

BatchPatchThread::BatchPatchThread(BatchPatch * patch) : _batchPatch(patch)
{
 _shutDown = false;
}

BatchPatchThread::~BatchPatchThread()
{
 terminate();
}

void BatchPatchThread::terminate()
{
 _shutDown = true;

 if (isAlive())
 {
 getThreadControl().join();
 }
}

void BatchPatchThread::doPatchRequest(const tars::PatchRequest & request, ServerObjectPtr server)
{
 NODE_LOG("patchPro")->debug() << FILE_FUN
 << request.appname + "." + request.servername + "_" + request.nodename << "|"
 << request.groupname << "|"
 << request.version << "|"
 << request.user << "|"
 << request.servertype << "|"
 << request.patchobj << "|"
 << request.md5 << "|"
 << request.ostype << endl;


 //设置发布状态
 try
 {
 std::string sError;

 if (!server)
 {
 NODE_LOG("patchPro")->error() <<FILE_FUN<< request.appname + "." + request.servername << "|" << request.md5 << "|get server object fault:" << sError << endl;
 return ; 
 }

 //查看本地硬盘是否已经存在该文件
 NODE_LOG("patchPro")->debug() <<FILE_FUN<< request.appname + "." + request.servername << "|" << request.md5 << "|patch begin" << endl;

 CommandPatch command(server, _downloadPath, request);
 if (command.canExecute(sError) != ServerCommand::EXECUTABLE)
 {
 NODE_LOG("patchPro")->error() << FILE_FUN<< request.appname + "." + request.servername << "|" << request.md5 << "|canExecute error:" << sError << endl;
 return ;
 }

 if (command.execute(sError) == 0)
 {
 NODE_LOG("patchPro")->debug() <<FILE_FUN<< request.appname + "." + request.servername << "|" << request.md5 << "|patch succ" << endl;
 }
 else
 {
 NODE_LOG("patchPro")->error() <<FILE_FUN<< request.appname + "." + request.servername << "|" << request.md5 << "|patch fault:" << sError << endl;
 }
 }
 catch (exception & e)
 {
 NODE_LOG("patchPro")->error() << FILE_FUN<< request.appname + "." + request.servername << "|" << request.md5 << "|Exception:" << e.what() << endl;
 }
 catch (...)
 {
 NODE_LOG("patchPro")->error() <<FILE_FUN<< request.appname + "." + request.servername << "|" << request.md5 << "|Unknown Exception" << endl;
 }
}

void BatchPatchThread::run()
{
 while (!_shutDown)
 {
 try
 {
 
 pair<tars::PatchRequest,ServerObjectPtr> item;
 if (_batchPatch->pop_front(item))
 {
 if (!(item.second))
 {
 NODE_LOG("patchPro")->error() <<FILE_FUN<< " server error" << endl;
 throw TC_Exception("server error");
 }
 doPatchRequest(item.first,item.second);
 }
 else
 {
 _batchPatch->timedWait(2000);
 }
 }
 catch (exception& e)
 {
 NODE_LOG("patchPro")->error()<<FILE_FUN<<"catch exception|"<<e.what()<<endl;
 }
 catch (...)
 {
 NODE_LOG("patchPro")->error()<<FILE_FUN<<"catch unkown exception"<<endl;
 }
 }
}