mirrors/TarsCloud_TarsCpp
1
0
Fork 0
mirror of https://gitee.com/TarsCloud/TarsCpp.git synced 2024-12-22 22:16:38 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #3 from TarsCloud/master

Browse Source 
merge
This commit is contained in:
wincsb 2020-04-24 18:57:04 +08:00 committed by GitHub
commit 99974b479c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
27 changed files with 5755 additions and 386 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -21,7 +21,7 @@ linux kernel: | 2.6.18 and above
gcc: | 4.1.2 and above glibc-devel
bison tool: | 2.5 and above
flex tool: | 2.5 and above
cmake: | 2.8.8 and above
cmake: | 3.2 and above
mysql: | 4.1.17 and above
Compile and install
@ -33,4 +33,4 @@ make
make install
```
Detailed [reference](https://tarscloud.github.io/TarsDocs/an-li/tarscpp/README.en.html)
Detailed [reference](https://tarscloud.github.io/TarsDocs_en/)

4
README.zh.md
View File

@ -21,7 +21,7 @@ linux内核版本: | 2.6.18及以上版本
gcc版本: | 4.1.2及以上版本、glibc-devel
bison工具版本: | 2.5及以上版本
flex工具版本: | 2.5及以上版本
cmake版本： | 2.8.8及以上版本
cmake版本： | 3.2及以上版本
mysql版本： | 4.1.17及以上版本
编译和安装
@ -33,4 +33,4 @@ make
make install
```
详细安装参见[案例文档](https://tarscloud.github.io/TarsDocs/an-li/tarscpp/)
详细安装参见[案例文档](https://tarscloud.github.io/TarsDocs/)

96
servant/libservant/AdapterProxy.cpp
View File

@ -143,7 +143,7 @@ void AdapterProxy::initStatHead()
//bool AdapterProxy::invoke_sync(ReqMessage * msg)
//{
// if(!_trans->hasConnected()) {
// TLOGTARS("[TARS][AdapterProxy::invoke_sync " << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
// TLOGTARS("[AdapterProxy::invoke_sync " << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
// return false;
// }
//
@ -182,14 +182,14 @@ int AdapterProxy::invoke_connection_serial(ReqMessage * msg)
if(!_requestMsg && _timeoutQueue->sendListEmpty() && _trans->sendRequest(msg->sReqData) != Transceiver::eRetError)
{
TLOGTARS("[TARS][AdapterProxy::invoke push (send) obj: " << _objectProxy->name() << ", desc:" << _endpoint.desc() << ", id: " << msg->request.iRequestId << endl);
TLOGTARS("[AdapterProxy::invoke push (send) obj: " << _objectProxy->name() << ", desc:" << _endpoint.desc() << ", id: " << msg->request.iRequestId << endl);
_requestMsg = msg;
bool bFlag = _timeoutQueue->push(msg, msg->request.iRequestId, msg->request.iTimeout + msg->iBeginTime);
if (!bFlag)
{
TLOGERROR("[TARS][AdapterProxy::invoke fail1 : insert timeout queue fail,queue size:" << _timeoutQueue->size() << ",id: " << msg->request.iRequestId << "," << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
TLOGERROR("[AdapterProxy::invoke fail1 : insert timeout queue fail,queue size:" << _timeoutQueue->size() << ",id: " << msg->request.iRequestId << "," << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
msg->eStatus = ReqMessage::REQ_EXC;
finishInvoke(msg);
@ -197,13 +197,13 @@ int AdapterProxy::invoke_connection_serial(ReqMessage * msg)
}
else
{
TLOGTARS("[TARS][AdapterProxy::invoke push (no send) " << _objectProxy->name() << ", " << _endpoint.desc() << ",id " << msg->request.iRequestId << endl);
TLOGTARS("[AdapterProxy::invoke push (no send) " << _objectProxy->name() << ", " << _endpoint.desc() << ",id " << msg->request.iRequestId << endl);
//之前还没有数据没发送 或者 请求发送失败了, 进队列
bool bFlag = _timeoutQueue->push(msg, msg->request.iRequestId, msg->request.iTimeout + msg->iBeginTime, false);
if (!bFlag)
{
TLOGERROR("[TARS][AdapterProxy::invoke fail2 : insert timeout queue fail,queue size:" << _timeoutQueue->size() << ", id: " << msg->request.iRequestId << ", " << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
TLOGERROR("[AdapterProxy::invoke fail2 : insert timeout queue fail,queue size:" << _timeoutQueue->size() << ", id: " << msg->request.iRequestId << ", " << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
msg->eStatus = ReqMessage::REQ_EXC;
finishInvoke(msg);
@ -217,12 +217,12 @@ int AdapterProxy::invoke_connection_parallel(ReqMessage * msg)
{
msg->sReqData->setBuffer(_objectProxy->getProxyProtocol().requestFunc(msg->request, _trans.get()));
// TLOGERROR("[TARS][AdapterProxy::invoke insert timeout queue fail, queue size:" << _timeoutQueue->size() << ", id:" << msg->request.iRequestId << ", obj:" <<_objectProxy->name() << ", desc:" << _endpoint.desc() <<endl);
// TLOGERROR("[AdapterProxy::invoke insert timeout queue fail, queue size:" << _timeoutQueue->size() << ", id:" << msg->request.iRequestId << ", obj:" <<_objectProxy->name() << ", desc:" << _endpoint.desc() <<endl);
//链表是空的, 则直接发送当前这条数据, 如果链表非空或者发送失败了, 则放到队列中, 等待下次发送
if(_timeoutQueue->sendListEmpty() && _trans->sendRequest(msg->sReqData) != Transceiver::eRetError)
{
TLOGTARS("[TARS][AdapterProxy::invoke push (send) obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",id:" << msg->request.iRequestId << endl);
TLOGTARS("[AdapterProxy::invoke push (send) obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",id:" << msg->request.iRequestId << endl);
//请求发送成功了，单向调用直接返回
if(msg->eType == ReqMessage::ONE_WAY)
@ -240,7 +240,7 @@ int AdapterProxy::invoke_connection_parallel(ReqMessage * msg)
bool bFlag = _timeoutQueue->push(msg, msg->request.iRequestId, msg->request.iTimeout + msg->iBeginTime);
if(!bFlag)
{
TLOGERROR("[TARS][AdapterProxy::invoke fail1: insert timeout queue fail, queue size:" << _timeoutQueue->size() << ", id:" << msg->request.iRequestId << ", obj:" <<_objectProxy->name() << ", desc:" << _endpoint.desc() <<endl);
TLOGERROR("[AdapterProxy::invoke fail1: insert timeout queue fail, queue size:" << _timeoutQueue->size() << ", id:" << msg->request.iRequestId << ", obj:" <<_objectProxy->name() << ", desc:" << _endpoint.desc() <<endl);
msg->eStatus = ReqMessage::REQ_EXC;
finishInvoke(msg);
@ -250,13 +250,13 @@ int AdapterProxy::invoke_connection_parallel(ReqMessage * msg)
}
else
{
TLOGTARS("[TARS][AdapterProxy::invoke push (no send) " << _objectProxy->name() << ", " << _endpoint.desc() << ",id:" << msg->request.iRequestId <<endl);
TLOGTARS("[AdapterProxy::invoke push (no send) " << _objectProxy->name() << ", " << _endpoint.desc() << ",id:" << msg->request.iRequestId <<endl);
//请求发送失败了
bool bFlag = _timeoutQueue->push(msg,msg->request.iRequestId, msg->request.iTimeout+msg->iBeginTime, false);
if(!bFlag)
{
TLOGERROR("[TARS][AdapterProxy::invoke fail2: insert timeout queue fail, queue size:" << _timeoutQueue->size() << ", id:" << msg->request.iRequestId << ", obj:" <<_objectProxy->name() << ", desc:" << _endpoint.desc() <<endl);
TLOGERROR("[AdapterProxy::invoke fail2: insert timeout queue fail, queue size:" << _timeoutQueue->size() << ", id:" << msg->request.iRequestId << ", obj:" <<_objectProxy->name() << ", desc:" << _endpoint.desc() <<endl);
msg->eStatus = ReqMessage::REQ_EXC;
@ -271,12 +271,12 @@ int AdapterProxy::invoke(ReqMessage * msg)
{
assert(_trans != NULL);
TLOGTARS("[TARS][AdapterProxy::invoke " << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
TLOGTARS("[AdapterProxy::invoke " << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
//未发链表有长度限制
if (_timeoutQueue->getSendListSize() >= _noSendQueueLimit)
{
TLOGERROR("[TARS][AdapterProxy::invoke fail,ReqInfoQueue.size>" << _noSendQueueLimit << "," << _objectProxy->name() << "," << _endpoint.desc() << "]" << endl);
TLOGERROR("[AdapterProxy::invoke fail,ReqInfoQueue.size>" << _noSendQueueLimit << "," << _objectProxy->name() << "," << _endpoint.desc() << "]" << endl);
msg->eStatus = ReqMessage::REQ_EXC;
finishInvoke(msg);
@ -305,7 +305,7 @@ int AdapterProxy::invoke(ReqMessage * msg)
// //连接连上 buffer不为空 发送数据成功
// if (_timeoutQueue->sendListEmpty() && _trans->sendRequest(msg->sReqData) != Transceiver::eRetError)
// {
// TLOGTARS("[TARS][AdapterProxy::invoke push (send) obj: " << _objectProxy->name() << ", desc:" << _endpoint.desc() << ", id: " << msg->request.iRequestId << endl);
// TLOGTARS("[AdapterProxy::invoke push (send) obj: " << _objectProxy->name() << ", desc:" << _endpoint.desc() << ", id: " << msg->request.iRequestId << endl);
//
// //请求发送成功了 处理采样
// //这个请求发送成功了。单向调用直接返回
@ -320,7 +320,7 @@ int AdapterProxy::invoke(ReqMessage * msg)
// bool bFlag = _timeoutQueue->push(msg, msg->request.iRequestId, msg->request.iTimeout + msg->iBeginTime);
// if (!bFlag)
// {
// TLOGERROR("[TARS][AdapterProxy::invoke fail1 : insert timeout queue fail,queue size:" << _timeoutQueue->size() << ",id: " << msg->request.iRequestId << "," << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
// TLOGERROR("[AdapterProxy::invoke fail1 : insert timeout queue fail,queue size:" << _timeoutQueue->size() << ",id: " << msg->request.iRequestId << "," << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
// msg->eStatus = ReqMessage::REQ_EXC;
//
// finishInvoke(msg);
@ -328,13 +328,13 @@ int AdapterProxy::invoke(ReqMessage * msg)
// }
// else
// {
// TLOGTARS("[TARS][AdapterProxy::invoke push (no send) " << _objectProxy->name() << ", " << _endpoint.desc() << ",id " << msg->request.iRequestId << endl);
// TLOGTARS("[AdapterProxy::invoke push (no send) " << _objectProxy->name() << ", " << _endpoint.desc() << ",id " << msg->request.iRequestId << endl);
//
// //之前还没有数据没发送 或者 请求发送失败了, 进队列
// bool bFlag = _timeoutQueue->push(msg, msg->request.iRequestId, msg->request.iTimeout + msg->iBeginTime, false);
// if (!bFlag)
// {
// TLOGERROR("[TARS][AdapterProxy::invoke fail2 : insert timeout queue fail,queue size:" << _timeoutQueue->size() << ", id: " << msg->request.iRequestId << ", " << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
// TLOGERROR("[AdapterProxy::invoke fail2 : insert timeout queue fail,queue size:" << _timeoutQueue->size() << ", id: " << msg->request.iRequestId << ", " << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
// msg->eStatus = ReqMessage::REQ_EXC;
//
// finishInvoke(msg);
@ -359,7 +359,7 @@ void AdapterProxy::doInvoke_serial()
//发送失败 返回
if (iRet == Transceiver::eRetError)
{
TLOGTARS("[TARS][AdapterProxy::doInvoke sendRequest failed, obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",id:" << msg->request.iRequestId << ", ret:" << iRet << endl);
TLOGTARS("[AdapterProxy::doInvoke sendRequest failed, obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",id:" << msg->request.iRequestId << ", ret:" << iRet << endl);
return;
}
@ -383,7 +383,7 @@ void AdapterProxy::doInvoke_parallel()
//发送失败 返回
if(iRet == Transceiver::eRetError)
{
TLOGTARS("[TARS][AdapterProxy::doInvoke sendRequest failed, obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",id:" << msg->request.iRequestId << ", ret:" << iRet << endl);
TLOGTARS("[AdapterProxy::doInvoke sendRequest failed, obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",id:" << msg->request.iRequestId << ", ret:" << iRet << endl);
return;
}
@ -423,7 +423,7 @@ void AdapterProxy::doInvoke(bool initInvoke)
void AdapterProxy::finishInvoke(bool bFail)
{
TLOGTARS("[TARS][AdapterProxy::finishInvoke(bool) obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",bFail:" << bFail << endl);
TLOGTARS("[AdapterProxy::finishInvoke(bool) obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",bFail:" << bFail << endl);
time_t now = TNOW;
@ -452,12 +452,12 @@ void AdapterProxy::finishInvoke(bool bFail)
_connExc = false;
TLOGTARS("[TARS][AdapterProxy::finishInvoke(bool), objname:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",retry ok" << endl);
TLOGTARS("[AdapterProxy::finishInvoke(bool), objname:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",retry ok" << endl);
}
else
{
//结点已经屏蔽 过来失败的包不用处理
TLOGTARS("[TARS][AdapterProxy::finishInvoke(bool), objname:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",retry fail" << endl);
TLOGTARS("[AdapterProxy::finishInvoke(bool), objname:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",retry fail" << endl);
}
return;
}
@ -487,7 +487,7 @@ void AdapterProxy::finishInvoke(bool bFail)
_activeStatus = false;
_nextRetryTime = TNOW + _objectProxy->checkTimeoutInfo().tryTimeInterval;
TLOGERROR("[TARS][AdapterProxy::finishInvoke(bool) objname:"<< _objectProxy->name()
TLOGERROR("[AdapterProxy::finishInvoke(bool) objname:"<< _objectProxy->name()
<< ",desc:" << _endpoint.desc()
<< ",disable frequenceFail,freqtimeout:" << _frequenceFailInvoke
<< ",timeout:"<< _timeoutInvoke
@ -508,7 +508,7 @@ void AdapterProxy::finishInvoke(bool bFail)
if(bFail && _timeoutInvoke >= info.minTimeoutInvoke && _timeoutInvoke >= info.radio * _totalInvoke)
{
setInactive();
TLOGERROR("[TARS][AdapterProxy::finishInvoke(bool) objname" << _objectProxy->name()
TLOGERROR("[AdapterProxy::finishInvoke(bool) objname" << _objectProxy->name()
<< ",desc:" << _endpoint.desc()
<< ",disable radioFail,freqtimeout:" << _frequenceFailInvoke
<< ",timeout:"<< _timeoutInvoke
@ -539,7 +539,7 @@ bool AdapterProxy::checkActive(bool bForceConnect, bool onlyCheck)
{
time_t now = TNOW;
TLOGTARS("[TARS][AdapterProxy::checkActive objname:" << _objectProxy->name()
TLOGTARS("[AdapterProxy::checkActive objname:" << _objectProxy->name()
<< ",desc:" << _endpoint.desc()
<< ",_activeStatus:" << (_activeStatus ? "enable" : "disable")
<< (bForceConnect? ",forceConnect" : "")
@ -578,7 +578,7 @@ bool AdapterProxy::checkActive(bool bForceConnect, bool onlyCheck)
_trans->close();
TLOGERROR("[TARS][AdapterProxy::checkActive connect objname:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ", ex:" << ex.what() << endl);
TLOGERROR("[AdapterProxy::checkActive connect objname:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ", ex:" << ex.what() << endl);
}
}
@ -588,7 +588,7 @@ bool AdapterProxy::checkActive(bool bForceConnect, bool onlyCheck)
//失效且没有到下次重试时间, 直接返回不可用
if((!_activeStatus) && (now < _nextRetryTime) )
{
TLOGTARS("[TARS][AdapterProxy::checkActive,not reach retry time ,objname:" << _objectProxy->name() << ",desc:" << _endpoint.desc() <<endl);
TLOGTARS("[AdapterProxy::checkActive,not reach retry time ,objname:" << _objectProxy->name() << ",desc:" << _endpoint.desc() <<endl);
return false;
}
@ -610,7 +610,7 @@ bool AdapterProxy::checkActive(bool bForceConnect, bool onlyCheck)
_trans->close();
TLOGERROR("[TARS][AdapterProxy::checkActive connect objname:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ", ex:" << ex.what() << endl);
TLOGERROR("[AdapterProxy::checkActive connect objname:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ", ex:" << ex.what() << endl);
}
}
@ -623,7 +623,7 @@ void AdapterProxy::setConTimeout(bool bConTimeout)
{
if(bConTimeout != _connTimeout)
{
TLOGERROR("[TARS][AdapterProxy::setConTimeout desc:"<< _endpoint.desc() << " connect timeout status is:" << bConTimeout << endl);
TLOGERROR("[AdapterProxy::setConTimeout desc:"<< _endpoint.desc() << " connect timeout status is:" << bConTimeout << endl);
_connTimeout = bConTimeout;
if(_connTimeout)
{
@ -649,18 +649,18 @@ void AdapterProxy::setInactive()
_objectProxy->getServantProxy()->onSetInactive(_endpoint);
TLOGTARS("[TARS][AdapterProxy::setInactive, " << _objectProxy->name() << ", " << _endpoint.desc() << ", inactive]" << endl);
TLOGTARS("[AdapterProxy::setInactive, " << _objectProxy->name() << ", " << _endpoint.desc() << ", inactive]" << endl);
}
void AdapterProxy::finishInvoke_serial(shared_ptr<ResponsePacket> & rsp)
{
TLOGTARS("[TARS][AdapterProxy::finishInvoke(ResponsePacket), " << _objectProxy->name() << ", " << _endpoint.desc() << ", id:" << rsp->iRequestId << "]" << endl);
TLOGTARS("[AdapterProxy::finishInvoke(ResponsePacket), " << _objectProxy->name() << ", " << _endpoint.desc() << ", id:" << rsp->iRequestId << "]" << endl);
if (!_requestMsg)
{
if(_timeoutLogFlag)
{
TLOGERROR("[TARS][AdapterProxy::finishInvoke(ResponsePacket),"
TLOGERROR("[AdapterProxy::finishInvoke(ResponsePacket),"
<< _objectProxy->name()
<< ",get req-ptr NULL,may be timeout,id:"
<< rsp->iRequestId << ",desc:" << _endpoint.desc() << "]" << endl);
@ -693,7 +693,7 @@ void AdapterProxy::finishInvoke_serial(shared_ptr<ResponsePacket> & rsp)
void AdapterProxy::finishInvoke_parallel(shared_ptr<ResponsePacket> & rsp)
{
TLOGTARS("[TARS][AdapterProxy::finishInvoke(ResponsePacket), " << _objectProxy->name() << ", " << _endpoint.desc() << ", id:" << rsp->iRequestId << "]" << endl);
TLOGTARS("[AdapterProxy::finishInvoke(ResponsePacket), " << _objectProxy->name() << ", " << _endpoint.desc() << ", id:" << rsp->iRequestId << "]" << endl);
ReqMessage * msg = NULL;
@ -702,7 +702,7 @@ void AdapterProxy::finishInvoke_parallel(shared_ptr<ResponsePacket> & rsp)
{
if(!_objectProxy->getPushCallback())
{
TLOGERROR("[TARS][AdapterProxy::finishInvoke(ResponsePacket)，request id is 0, pushcallback is null, objname:" << _objectProxy->name()
TLOGERROR("[AdapterProxy::finishInvoke(ResponsePacket)，request id is 0, pushcallback is null, objname:" << _objectProxy->name()
<< ",desc:" << _endpoint.desc() << endl);
return;
}
@ -727,7 +727,7 @@ void AdapterProxy::finishInvoke_parallel(shared_ptr<ResponsePacket> & rsp)
{
if(_timeoutLogFlag)
{
TLOGERROR("[TARS][AdapterProxy::finishInvoke(ResponsePacket) obj:"<< _objectProxy->name() << ",get req-ptr NULL,may be timeout,id:" << rsp->iRequestId
TLOGERROR("[AdapterProxy::finishInvoke(ResponsePacket) obj:"<< _objectProxy->name() << ",get req-ptr NULL,may be timeout,id:" << rsp->iRequestId
<< ",desc:" << _endpoint.desc() << endl);
}
return ;
@ -745,7 +745,7 @@ void AdapterProxy::finishInvoke_parallel(shared_ptr<ResponsePacket> & rsp)
void AdapterProxy::finishInvoke(shared_ptr<ResponsePacket> & rsp)
{
TLOGTARS("[TARS][AdapterProxy::finishInvoke(ResponsePacket), " << _objectProxy->name() << ", " << _endpoint.desc() << ", id:" << rsp->iRequestId << "]" << endl);
TLOGTARS("[AdapterProxy::finishInvoke(ResponsePacket), " << _objectProxy->name() << ", " << _endpoint.desc() << ", id:" << rsp->iRequestId << "]" << endl);
if(_objectProxy->getServantProxy()->tars_connection_serial() > 0)
{
@ -765,7 +765,7 @@ void AdapterProxy::finishInvoke(shared_ptr<ResponsePacket> & rsp)
// //push callback is null
// if (!_objectProxy->getPushCallback())
// {
// TLOGERROR("[TARS][AdapterProxy::finishInvoke(ResponsePacket)， request id is 0, pushcallback is null, " << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
// TLOGERROR("[AdapterProxy::finishInvoke(ResponsePacket)， request id is 0, pushcallback is null, " << _objectProxy->name() << ", " << _endpoint.desc() << "]" << endl);
// return;
// }
// msg = new ReqMessage();
@ -789,7 +789,7 @@ void AdapterProxy::finishInvoke(shared_ptr<ResponsePacket> & rsp)
// {
// if (_timeoutLogFlag)
// {
// TLOGERROR("[TARS][AdapterProxy::finishInvoke(ResponsePacket),"
// TLOGERROR("[AdapterProxy::finishInvoke(ResponsePacket),"
// << _objectProxy->name()
// << ",get req-ptr NULL,may be timeout,id:"
// << rsp->iRequestId << ",desc:" << _endpoint.desc() << "]" << endl);
@ -811,7 +811,7 @@ void AdapterProxy::finishInvoke(ReqMessage * msg)
{
assert(msg->eStatus != ReqMessage::REQ_REQ);
TLOGTARS("[TARS][AdapterProxy::finishInvoke(ReqMessage) obj:" << _objectProxy->name() << ", desc:" << _endpoint.desc()
TLOGTARS("[AdapterProxy::finishInvoke(ReqMessage) obj:" << _objectProxy->name() << ", desc:" << _endpoint.desc()
<< " ,id:" << msg->response->iRequestId
<< ", status:" << msg->eStatus
<< ", ret: " << msg->response->iRet << endl);
@ -872,11 +872,11 @@ void AdapterProxy::finishInvoke(ReqMessage * msg)
}
catch (exception & e)
{
TLOGERROR("[TARS]AdapterProxy::finishInvoke(ReqMessage) exp:" << e.what() << " ,line:" << __LINE__ << endl);
TLOGERROR("AdapterProxy::finishInvoke(ReqMessage) exp:" << e.what() << " ,line:" << __LINE__ << endl);
}
catch (...)
{
TLOGERROR("[TARS]AdapterProxy::finishInvoke(ReqMessage) exp:unknown line:|" << __LINE__ << endl);
TLOGERROR("AdapterProxy::finishInvoke(ReqMessage) exp:unknown line:|" << __LINE__ << endl);
}
}
else
@ -898,7 +898,7 @@ void AdapterProxy::finishInvoke(ReqMessage * msg)
}
else
{
TLOGERROR("[TARS][AdapterProxy::finishInvoke(ReqMessage) coro parallel callback error,obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc()
TLOGERROR("[AdapterProxy::finishInvoke(ReqMessage) coro parallel callback error,obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc()
<< ",id:" << msg->response->iRequestId << endl);
delete msg;
msg = NULL;
@ -917,7 +917,7 @@ void AdapterProxy::doTimeout()
ReqMessage * msg;
while(_timeoutQueue->timeout(msg))
{
TLOGTARS("[TARS][AdapterProxy::doTimeout obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",id " << msg->request.iRequestId << endl);
TLOGTARS("[AdapterProxy::doTimeout obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << ",id " << msg->request.iRequestId << endl);
// assert(msg->eStatus == ReqMessage::REQ_REQ);
@ -955,7 +955,7 @@ void AdapterProxy::startTrack(ReqMessage * msg)
{
if(!_communicator->_traceManager)
{
TLOGTARS("[TARS]tracer info is null, just return" << endl);
TLOGTARS("tracer info is null, just return" << endl);
return;
}
@ -1022,7 +1022,7 @@ void AdapterProxy::stat(ReqMessage * msg)
{
return ;
}
TLOGTARS("[TARS]AdapterProxy::stat(ReqMessage) " << _objectProxy->name() << ", " << _endpoint.desc() << " ,id:" << msg->response->iRequestId << endl);
TLOGTARS("AdapterProxy::stat(ReqMessage) " << _objectProxy->name() << ", " << _endpoint.desc() << " ,id:" << msg->response->iRequestId << endl);
StatMicMsgBody body;
int64_t sptime = 0;
@ -1064,7 +1064,7 @@ void AdapterProxy::stat(ReqMessage * msg)
_statHead.displaySimple(os);
os << " ";
_statBody[msg->request.sFuncName].displaySimple(os);
TLOGTARS("[TARS][AdapterProxy::stat(ReqMessage) display:" << os.str() << endl);
TLOGTARS("[AdapterProxy::stat(ReqMessage) display:" << os.str() << endl);
}
}
@ -1091,7 +1091,7 @@ void AdapterProxy::merge(const StatMicMsgBody& inBody,StatMicMsgBody& outBody/*o
void AdapterProxy::mergeStat(map<StatMicMsgHead, StatMicMsgBody> & mStatMicMsg)
{
TLOGTARS("[TARS][AdapterProxy::doStat obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << endl);
TLOGTARS("[AdapterProxy::doStat obj:" << _objectProxy->name() << ",desc:" << _endpoint.desc() << endl);
for (const auto& kv : _statBody)
{
@ -1123,7 +1123,7 @@ void AdapterProxy::addConnExc(bool bExc)
if(!_connExc && _connExcCnt++ >= _objectProxy->checkTimeoutInfo().maxConnectExc)
{
TLOGERROR("[TARS][AdapterProxy::addConnExc desc:"<< _endpoint.desc() << ",connect exception status is true! (connect error)"<<endl);
TLOGERROR("[AdapterProxy::addConnExc desc:"<< _endpoint.desc() << ",connect exception status is true! (connect error)"<<endl);
setInactive();
_connExc = true;
@ -1133,7 +1133,7 @@ void AdapterProxy::addConnExc(bool bExc)
{
if(_connExc)
{
TLOGERROR("[TARS][AdapterProxy::addConnExc desc:"<< _endpoint.desc() << ",connect exception status is false!(connect ok)"<<endl);
TLOGERROR("[AdapterProxy::addConnExc desc:"<< _endpoint.desc() << ",connect exception status is false!(connect ok)"<<endl);
}
_connExc = false;

2
servant/libservant/AppProtocol.cpp
View File

@ -214,7 +214,7 @@ vector<char> ProxyProtocol::http2Request(RequestPacket& request, Transceiver *tr
if (request.iRequestId < 0)
{
TLOGERROR("[TARS]http2Request::Fatal submit error: " << session->getErrMsg() << endl);
TLOGERROR("http2Request::Fatal submit error: " << session->getErrMsg() << endl);
return vector<char>();
}

2
servant/libservant/Application.cpp
View File

@ -1264,7 +1264,7 @@ void Application::setAdapter(TC_EpollServer::BindAdapterPtr& adapter, const stri
shared_ptr<TC_OpenSSL::CTX> ctx = TC_OpenSSL::newCtx(ca, cert, key, verifyClient, ciphers);
if (!ctx) {
TLOGERROR("[TARS]load server ssl error, cert:" << cert << endl);
TLOGERROR("load server ssl error, cert:" << cert << endl);
exit(-1);
}

2
servant/libservant/AsyncProcThread.cpp
View File

@ -64,7 +64,7 @@ void AsyncProcThread::push_back(ReqMessage * msg)
else {
if(_msgQueue->size() >= _iQueueCap)
{
TLOGERROR("[TARS][AsyncProcThread::push_back] async_queue full:" << _msgQueue->size() << ">=" << _iQueueCap << endl);
TLOGERROR("[AsyncProcThread::push_back] async_queue full:" << _msgQueue->size() << ">=" << _iQueueCap << endl);
delete msg;
}
else

6
servant/libservant/Communicator.cpp
View File

@ -181,7 +181,7 @@ void Communicator::setProperty(TC_Config& conf, const string& domain/* = CONFIG_
shared_ptr<TC_OpenSSL::CTX> ctx = TC_OpenSSL::newCtx( data["ca"], data["cert"], data["key"], false, data["ciphers"]);
if(!ctx)
{
TLOGERROR("[TARS]load obj:" << auths[i] << ", ssl error, ca:" << data["ca"] << endl);
TLOGERROR("load obj:" << auths[i] << ", ssl error, ca:" << data["ca"] << endl);
exit(-1);
}
@ -219,7 +219,7 @@ void Communicator::initialize()
//set分组名不对时默认没有打开set分组
ClientConfig::SetOpen = false;
setProperty("enableset","n");
TLOGERROR( "[TARS][set division name error:" << ClientConfig::SetDivision << ", client failed to open set]" << endl);
TLOGERROR( "[set division name error:" << ClientConfig::SetDivision << ", client failed to open set]" << endl);
}
}
@ -264,7 +264,7 @@ void Communicator::initialize()
if(!_ctx)
{
TLOGERROR("[TARS]load client ssl error, ca:" << ca << endl);
TLOGERROR("load client ssl error, ca:" << ca << endl);
exit(-1);
}
}

32
servant/libservant/CommunicatorEpoll.cpp
View File

@ -149,7 +149,7 @@ void CommunicatorEpoll::handleInputImp(Transceiver * pTransceiver)
{
pTransceiver->close();
pTransceiver->getAdapterProxy()->addConnExc(true);
TLOGERROR("[TARS][CommunicatorEpoll::handleInputImp] connect error "
TLOGERROR("[CommunicatorEpoll::handleInputImp] connect error "
<< pTransceiver->getAdapterProxy()->endpoint().desc()
<< "," << pTransceiver->getAdapterProxy()->getObjProxy()->name()
<< ",_connExcCnt=" << pTransceiver->getAdapterProxy()->ConnExcCnt()
@ -174,7 +174,7 @@ void CommunicatorEpoll::handleOutputImp(Transceiver * pTransceiver)
{
pTransceiver->close();
pTransceiver->getAdapterProxy()->addConnExc(true);
TLOGERROR("[TARS][CommunicatorEpoll::handleOutputImp] connect error "
TLOGERROR("[CommunicatorEpoll::handleOutputImp] connect error "
<< pTransceiver->getAdapterProxy()->endpoint().desc()
<< "," << pTransceiver->getAdapterProxy()->getObjProxy()->name()
<< ",_connExcCnt=" << pTransceiver->getAdapterProxy()->ConnExcCnt()
@ -240,11 +240,11 @@ void CommunicatorEpoll::handle(FDInfo * pFDInfo, const epoll_event &ev)
}
catch(exception & e)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
}
catch(...)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle|"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle|"<<__LINE__<<endl);
}
if(++maxProcessCount > 1000)
@ -257,11 +257,11 @@ void CommunicatorEpoll::handle(FDInfo * pFDInfo, const epoll_event &ev)
}
catch(exception & e)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
}
catch(...)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle|"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle|"<<__LINE__<<endl);
}
}
else
@ -278,11 +278,11 @@ void CommunicatorEpoll::handle(FDInfo * pFDInfo, const epoll_event &ev)
}
catch(exception & e)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
}
catch(...)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle|"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle|"<<__LINE__<<endl);
}
}
@ -295,11 +295,11 @@ void CommunicatorEpoll::handle(FDInfo * pFDInfo, const epoll_event &ev)
}
catch(exception & e)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
}
catch(...)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle|"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle|"<<__LINE__<<endl);
}
}
@ -312,22 +312,22 @@ void CommunicatorEpoll::handle(FDInfo * pFDInfo, const epoll_event &ev)
}
catch(exception & e)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
}
catch(...)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle|"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle|"<<__LINE__<<endl);
}
}
}
}
catch(exception & e)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle exp:"<<e.what()<<" ,line:"<<__LINE__<<endl);
}
catch(...)
{
TLOGERROR("[TARS]CommunicatorEpoll::handle|"<<__LINE__<<endl);
TLOGERROR("CommunicatorEpoll::handle|"<<__LINE__<<endl);
}
}
@ -477,11 +477,11 @@ void CommunicatorEpoll::run()
}
catch (exception& e)
{
TLOGERROR("[TARS][CommunicatorEpoll:run exception:" << e.what() << "]" << endl);
TLOGERROR("[CommunicatorEpoll:run exception:" << e.what() << "]" << endl);
}
catch (...)
{
TLOGERROR("[TARS][CommunicatorEpoll:run exception.]" << endl);
TLOGERROR("[CommunicatorEpoll:run exception.]" << endl);
}
}
}

42
servant/libservant/CoroutineScheduler.cpp
View File

@ -289,11 +289,11 @@ void CoroutineInfo::corotineEntry(transfer_t tf)
}
catch(std::exception &ex)
{
TLOGERROR("[TARS][CoroutineInfo::corotineEntry exception:" << ex.what() << endl);
TLOGERROR("[CoroutineInfo::corotineEntry exception:" << ex.what() << endl);
}
catch(...)
{
TLOGERROR("[TARS][CoroutineInfo::corotineEntry unknown exception." << endl);
TLOGERROR("[CoroutineInfo::corotineEntry unknown exception." << endl);
}
}
@ -309,11 +309,11 @@ void CoroutineInfo::corotineProc(void * args, transfer_t t)
}
catch(std::exception &ex)
{
TLOGERROR("[TARS][CoroutineInfo::corotineProc exception:" << ex.what() << endl);
TLOGERROR("[CoroutineInfo::corotineProc exception:" << ex.what() << endl);
}
catch(...)
{
TLOGERROR("[TARS][CoroutineInfo::corotineProc unknown exception." << endl);
TLOGERROR("[CoroutineInfo::corotineProc unknown exception." << endl);
}
CoroutineScheduler* scheduler = coro->getScheduler();
@ -322,7 +322,7 @@ void CoroutineInfo::corotineProc(void * args, transfer_t t)
scheduler->switchCoro(&(scheduler->getMainCoroutine()));
// scheduler->switchCoro(coro, &(scheduler->getMainCoroutine()));
TLOGERROR("[TARS][CoroutineInfo::corotineProc no come." << endl);
TLOGERROR("[CoroutineInfo::corotineProc no come." << endl);
}
//////////////////////////////////////////////////////////////
@ -357,7 +357,7 @@ void CoroutineScheduler::init(uint32_t iPoolSize, size_t iStackSize)
{
if(iPoolSize <= 0)
{
TLOGERROR("[TARS][[CoroutineScheduler::init iPoolSize <= 0." << endl);
TLOGERROR("[CoroutineScheduler::init iPoolSize <= 0." << endl);
return ;
}
@ -396,7 +396,7 @@ void CoroutineScheduler::init(uint32_t iPoolSize, size_t iStackSize)
// int ret = _alloc.allocate(s_ctx, iStackSize);
// if(ret != 0)
// {
// TLOGERROR("[TARS][CoroutineScheduler::init iPoolSize:" << iPoolSize << "|iStackSize:" << iStackSize << "|i:" << i << endl);
// TLOGERROR("[CoroutineScheduler::init iPoolSize:" << iPoolSize << "|iStackSize:" << iStackSize << "|i:" << i << endl);
// delete coro;
// coro = NULL;
@ -429,7 +429,7 @@ void CoroutineScheduler::init(uint32_t iPoolSize, size_t iStackSize)
_currentCoro = &_mainCoro;
TLOGDEBUG("[TARS][CoroutineScheduler::init iPoolSize:" << _poolSize << "|iCurrentSize:" << _currentSize << "|iStackSize:" << _stackSize << endl);
TLOGDEBUG("[CoroutineScheduler::init iPoolSize:" << _poolSize << "|iCurrentSize:" << _currentSize << "|iStackSize:" << _stackSize << endl);
}
int CoroutineScheduler::increaseCoroPoolSize()
@ -437,7 +437,7 @@ int CoroutineScheduler::increaseCoroPoolSize()
int iInc = ((_poolSize - _currentSize) > 100) ? 100 : (_poolSize - _currentSize);
if(iInc <= 0)
{
TLOGERROR("[TARS][CoroutineScheduler::increaseCoroPoolSize full iPoolSize:" << _poolSize << "|iCurrentSize:" << _currentSize << endl);
TLOGERROR("[CoroutineScheduler::increaseCoroPoolSize full iPoolSize:" << _poolSize << "|iCurrentSize:" << _currentSize << endl);
return -1;
}
@ -453,7 +453,7 @@ int CoroutineScheduler::increaseCoroPoolSize()
// int ret = _alloc.allocate(s_ctx, _stackSize);
// if(ret != 0)
// {
// TLOGERROR("[TARS][CoroutineScheduler::increaseCoroPoolSize iPoolSize:" << _poolSize << "|iStackSize:" << _stackSize << "|i:" << i << endl);
// TLOGERROR("[CoroutineScheduler::increaseCoroPoolSize iPoolSize:" << _poolSize << "|iStackSize:" << _stackSize << "|i:" << i << endl);
// delete coro;
// coro = NULL;
@ -473,7 +473,7 @@ int CoroutineScheduler::increaseCoroPoolSize()
if(iSucc == 0)
{
TLOGERROR("[TARS][CoroutineScheduler::increaseCoroPoolSize cannot create iInc:" << iInc << "|iPoolSize:" << _poolSize << endl);
TLOGERROR("[CoroutineScheduler::increaseCoroPoolSize cannot create iInc:" << iInc << "|iPoolSize:" << _poolSize << endl);
return -1;
}
@ -792,7 +792,7 @@ void CoroutineScheduler::moveToActive(CoroutineInfo *coro, bool bFlag)
}
else
{
TLOGERROR("[TARS][CoroutineScheduler::moveToActive ERROR|iCoroId:" << coro->getUid() << "|tyep:" << coro->getStatus() << endl);
TLOGERROR("[CoroutineScheduler::moveToActive ERROR|iCoroId:" << coro->getUid() << "|tyep:" << coro->getStatus() << endl);
}
}
@ -806,7 +806,7 @@ void CoroutineScheduler::moveToAvail(CoroutineInfo *coro)
}
else
{
TLOGERROR("[TARS][CoroutineScheduler::moveToAvail ERROR:|iCoroId:" << coro->getUid() << "|tyep:" << coro->getStatus() << endl);
TLOGERROR("[CoroutineScheduler::moveToAvail ERROR:|iCoroId:" << coro->getUid() << "|tyep:" << coro->getStatus() << endl);
}
}
@ -826,7 +826,7 @@ void CoroutineScheduler::moveToInactive(CoroutineInfo *coro)
}
else
{
TLOGERROR("[TARS][CoroutineScheduler::moveToInactive ERROR|iCoroId:" << coro->getUid() << "|tyep:" << coro->getStatus() << endl);
TLOGERROR("[CoroutineScheduler::moveToInactive ERROR|iCoroId:" << coro->getUid() << "|tyep:" << coro->getStatus() << endl);
}
}
@ -846,7 +846,7 @@ void CoroutineScheduler::moveToTimeout(CoroutineInfo *coro)
}
else
{
TLOGERROR("[TARS][CoroutineScheduler::moveToTimeout ERROR|iCoroId:" << coro->getUid() << "|tyep:" << coro->getStatus() << endl);
TLOGERROR("[CoroutineScheduler::moveToTimeout ERROR|iCoroId:" << coro->getUid() << "|tyep:" << coro->getStatus() << endl);
}
}
@ -878,7 +878,7 @@ void CoroutineScheduler::moveToFreeList(CoroutineInfo *coro)
}
else
{
TLOGERROR("[TARS][CoroutineScheduler::moveToFreeList ERROR: already free|iCoroId:" << coro->getUid() << "|tyep:" << coro->getStatus() << endl);
TLOGERROR("[CoroutineScheduler::moveToFreeList ERROR: already free|iCoroId:" << coro->getUid() << "|tyep:" << coro->getStatus() << endl);
}
}
@ -970,11 +970,11 @@ void Coroutine::coroEntry(Coroutine *pCoro)
}
catch(exception &ex)
{
TLOGERROR("[TARS][[Coroutine::coroEntry exception:" << ex.what() << "]" << endl);
TLOGERROR("[Coroutine::coroEntry exception:" << ex.what() << "]" << endl);
}
catch(...)
{
TLOGERROR("[TARS][[Coroutine::coroEntry unknown exception]" << endl);
TLOGERROR("[Coroutine::coroEntry unknown exception]" << endl);
}
}
@ -986,7 +986,7 @@ uint32_t Coroutine::createCoroutine(const std::function<void ()> &coroFunc)
}
else
{
TLOGERROR("[TARS][[Coroutine::createCoroutine coro sched no init]" << endl);
TLOGERROR("[Coroutine::createCoroutine coro sched no init]" << endl);
}
return -1;
}
@ -999,7 +999,7 @@ void Coroutine::yield()
}
else
{
throw CoroutineException("[TARS][[Coroutine::yield coro sched no init]");
throw CoroutineException("[Coroutine::yield coro sched no init]");
}
}
@ -1011,7 +1011,7 @@ void Coroutine::Sleep(int iSleepTime)
}
else
{
throw CoroutineException("[TARS][[Coroutine::yield coro sched no init]");
throw CoroutineException("[Coroutine::yield coro sched no init]");
}
}

4
servant/libservant/Current.cpp
View File

@ -335,7 +335,7 @@ void Current::sendResponse(int iRet, const vector<char> &buffer, const map<stri
response.context = _responseContext;
response.iRet = iRet;
TLOGTARS("[TARS]Current::sendResponse :"
TLOGTARS("Current::sendResponse :"
<< response.iMessageType << "|"
<< _request.sServantName << "|"
<< _request.sFuncName << "|"
@ -377,7 +377,7 @@ void Current::sendResponse(int iRet, const vector<char> &buffer, const map<stri
response.status[ServantProxy::STATUS_RESULT_DESC] = sResultDesc;
}
TLOGTARS("[TARS]Current::sendResponse :"
TLOGTARS("Current::sendResponse :"
<< response.iMessageType << "|"
<< _request.sServantName << "|"
<< _request.sFuncName << "|"

78
servant/libservant/EndpointManager.cpp
View File

@ -51,7 +51,7 @@ QueryEpBase::QueryEpBase(Communicator * pComm, bool bFirstNetThread,bool bInterf
void QueryEpBase::callback_findObjectById4All(Int32 ret, const vector<EndpointF>& activeEp, const vector<EndpointF>& inactiveEp)
{
TLOGTARS("[TARS][callback_findObjectById4All _objName:" << _objName << "|ret:" << ret
TLOGTARS("[callback_findObjectById4All _objName:" << _objName << "|ret:" << ret
<< ",active:" << activeEp.size()
<< ",inactive:" << inactiveEp.size() << "]" << endl);
@ -60,14 +60,14 @@ void QueryEpBase::callback_findObjectById4All(Int32 ret, const vector<EndpointF>
void QueryEpBase::callback_findObjectById4All_exception(Int32 ret)
{
TLOGERROR("[TARS][callback_findObjectById4All_exception _objName:" << _objName << "|ret:" << ret << "]" << endl);
TLOGERROR("[callback_findObjectById4All_exception _objName:" << _objName << "|ret:" << ret << "]" << endl);
doEndpointsExp(ret);
}
void QueryEpBase::callback_findObjectById4Any(Int32 ret, const vector<EndpointF>& activeEp, const vector<EndpointF>& inactiveEp)
{
TLOGTARS("[TARS][callback_findObjectById4Any _objName:" << _objName << "|ret:" << ret
TLOGTARS("[callback_findObjectById4Any _objName:" << _objName << "|ret:" << ret
<< ",active:" << activeEp.size()
<< ",inactive:" << inactiveEp.size() << "]" << endl);
@ -76,14 +76,14 @@ void QueryEpBase::callback_findObjectById4Any(Int32 ret, const vector<EndpointF>
void QueryEpBase::callback_findObjectById4Any_exception(Int32 ret)
{
TLOGERROR("[TARS][callback_findObjectById4Any_exception _objName:" << _objName << "|ret:" << ret << "]" << endl);
TLOGERROR("[callback_findObjectById4Any_exception _objName:" << _objName << "|ret:" << ret << "]" << endl);
doEndpointsExp(ret);
}
void QueryEpBase::callback_findObjectByIdInSameGroup(Int32 ret, const vector<EndpointF>& activeEp, const vector<EndpointF>& inactiveEp)
{
TLOGTARS("[TARS][callback_findObjectByIdInSameGroup _objName:" << _objName << "|ret:"<<ret
TLOGTARS("[callback_findObjectByIdInSameGroup _objName:" << _objName << "|ret:"<<ret
<< ",active:" << activeEp.size()
<< ",inactive:" << inactiveEp.size() << "]" << endl);
@ -92,14 +92,14 @@ void QueryEpBase::callback_findObjectByIdInSameGroup(Int32 ret, const vector<End
void QueryEpBase::callback_findObjectByIdInSameGroup_exception(Int32 ret)
{
TLOGERROR("[TARS][callback_findObjectByIdInSameGroup_exception _objName:" << _objName << "|ret:" << ret << "]" << endl);
TLOGERROR("[callback_findObjectByIdInSameGroup_exception _objName:" << _objName << "|ret:" << ret << "]" << endl);
doEndpointsExp(ret);
}
void QueryEpBase::callback_findObjectByIdInSameSet( Int32 ret, const vector<EndpointF> &activeEp, const vector<EndpointF> & inactiveEp)
{
TLOGTARS("[TARS][callback_findObjectByIdInSameSet _objName:" << _objName << "|ret:" << ret
TLOGTARS("[callback_findObjectByIdInSameSet _objName:" << _objName << "|ret:" << ret
<< ",active:" << activeEp.size()
<< ",inactive:" << inactiveEp.size() << "]" << endl);
@ -108,14 +108,14 @@ void QueryEpBase::callback_findObjectByIdInSameSet( Int32 ret, const vector<Endp
void QueryEpBase::callback_findObjectByIdInSameSet_exception( Int32 ret)
{
TLOGERROR("[TARS][callback_findObjectByIdInSameSet_exception _objName:" << _objName << "|ret:" << ret << "]" << endl);
TLOGERROR("[callback_findObjectByIdInSameSet_exception _objName:" << _objName << "|ret:" << ret << "]" << endl);
doEndpointsExp(ret);
}
void QueryEpBase::callback_findObjectByIdInSameStation( Int32 ret, const vector<EndpointF> &activeEp, const vector<EndpointF> &inactiveEp)
{
TLOGTARS("[TARS][callback_findObjectByIdInSameStation _objName:" << _objName << "|ret:" << ret
TLOGTARS("[callback_findObjectByIdInSameStation _objName:" << _objName << "|ret:" << ret
<< ",active:" << activeEp.size()
<< ",inactive:" << inactiveEp.size() << "]" << endl);
@ -124,7 +124,7 @@ void QueryEpBase::callback_findObjectByIdInSameStation( Int32 ret, const vector<
void QueryEpBase::callback_findObjectByIdInSameStation_exception( Int32 ret)
{
TLOGERROR("[TARS][callback_findObjectByIdInSameStation_exception _objName:" << _objName << "|ret:" << ret << "]" << endl);
TLOGERROR("[callback_findObjectByIdInSameStation_exception _objName:" << _objName << "|ret:" << ret << "]" << endl);
doEndpointsExp(ret);
}
@ -138,7 +138,7 @@ int QueryEpBase::setLocatorPrx(QueryFPrx prx)
bool QueryEpBase::init(const string & sObjName,const string & sLocator,const string& setName)
{
TLOGTARS("[TARS][QueryEpBase::init sObjName:" << sObjName << ",sLocator:" << sLocator << ",setName:" << setName << "]" << endl);
TLOGTARS("[QueryEpBase::init sObjName:" << sObjName << ",sLocator:" << sLocator << ",setName:" << setName << "]" << endl);
_locator = sLocator;
@ -186,7 +186,7 @@ void QueryEpBase::setObjName(const string & sObjName)
if(_locator.find_first_not_of('@') == string::npos)
{
TLOGERROR("[TARS][QueryEpBase::setObjName locator is not valid,_locator:" << _locator << "]" << endl);
TLOGERROR("[QueryEpBase::setObjName locator is not valid,_locator:" << _locator << "]" << endl);
throw TarsRegistryException("locator is not valid,_locator:" + _locator);
}
@ -342,7 +342,7 @@ void QueryEpBase::setEndpoints(const string & sEndpoints, set<EndpointInfo> & se
}
catch (...)
{
TLOGERROR("[TARS][QueryEpBase::setEndpoints parse error,objname:" << _objName << ",endpoint:" << vEndpoints[i] << "]" << endl);
TLOGERROR("[QueryEpBase::setEndpoints parse error,objname:" << _objName << ",endpoint:" << vEndpoints[i] << "]" << endl);
}
}
@ -387,7 +387,7 @@ void QueryEpBase::refreshReg(GetEndpointType type, const string & sName)
//一定时间不回调就算超时了
_requestTimeout = iNow + _timeoutInterval;
TLOGTARS("[TARS][QueryEpBase::refresh,"<<_objName<<"]"<<endl);
TLOGTARS("[QueryEpBase::refresh,"<<_objName<<"]"<<endl);
//判断是同步调用还是异步调用
//内部请求主控都是异步请求
@ -476,12 +476,12 @@ void QueryEpBase::refreshReg(GetEndpointType type, const string & sName)
}
catch(TC_Exception & ex)
{
TLOGERROR("[TARS]QueryEpBase::refreshReg obj:"<<_objName<<"exception:"<<ex.what()<<endl);
TLOGERROR("[QueryEpBase::refreshReg obj:"<<_objName<<"exception:"<<ex.what() << "]"<<endl);
doEndpointsExp(TARSSERVERUNKNOWNERR);
}
catch(...)
{
TLOGERROR("[TARS]QueryEpBase::refreshReg obj:"<<_objName<<"unknown exception:"<<endl);
TLOGERROR("[QueryEpBase::refreshReg obj:"<<_objName<<"unknown exception]" <<endl);
doEndpointsExp(TARSSERVERUNKNOWNERR);
}
}
@ -508,7 +508,7 @@ void QueryEpBase::doEndpoints(const vector<EndpointF>& activeEp, const vector<En
_refreshTime = iNow + _activeEmptyInterval;
//如果registry返回Active服务列表为空，不做更新
TLOGERROR("[TARS][QueryEpBase::doEndpoints, callback activeEps is empty,objname:"<< _objName << "]" << endl);
TLOGERROR("[QueryEpBase::doEndpoints, callback activeEps is empty,objname:"<< _objName << "]" << endl);
return;
}
else
@ -691,7 +691,7 @@ void QueryEpBase::setEndPointToCache(bool bInactive)
AppCache::getInstance()->set(objName,sEndpoints,sLocatorKey);
}
TLOGTARS("[TARS][setEndPointToCache,obj:" << _objName << ",invokeSetId:" << _invokeSetId << ",endpoint:" << sEndpoints << "]" << endl);
TLOGTARS("[setEndPointToCache,obj:" << _objName << ",invokeSetId:" << _invokeSetId << ",endpoint:" << sEndpoints << "]" << endl);
}
/////////////////////////////////////////////////////////////////////////////
@ -865,7 +865,7 @@ AdapterProxy * EndpointManager::getNextValidProxy(bool onlyCheck)
{
if (_activeProxys.empty())
{
TLOGERROR("[TARS][EndpointManager::getNextValidProxy activeEndpoints is empty][obj:"<<_objName<<"]" << endl);
TLOGERROR("[EndpointManager::getNextValidProxy activeEndpoints is empty][obj:"<<_objName<<"]" << endl);
return NULL;
}
@ -938,7 +938,7 @@ AdapterProxy* EndpointManager::getHashProxyForWeight(int64_t hashCode, bool bSta
{
if(_vRegProxys.empty())
{
TLOGERROR("[TARS][EndpointManager::getHashProxyForWeight _vRegProxys is empty], bStatic:" << bStatic << endl);
TLOGERROR("[EndpointManager::getHashProxyForWeight _vRegProxys is empty], bStatic:" << bStatic << endl);
return NULL;
}
@ -950,7 +950,7 @@ AdapterProxy* EndpointManager::getHashProxyForWeight(int64_t hashCode, bool bSta
int64_t iEnd = TNOWMS;
TLOGTARS("[TARS][EndpointManager::getHashProxyForWeight update bStatic:" << bStatic << "|_objName:" << _objName << "|timecost(ms):" << (iEnd - iBegin) << endl);
TLOGTARS("[EndpointManager::getHashProxyForWeight update bStatic:" << bStatic << "|_objName:" << _objName << "|timecost(ms):" << (iEnd - iBegin) << endl);
}
if(vRouterCache.size() > 0)
@ -979,7 +979,7 @@ AdapterProxy* EndpointManager::getHashProxyForWeight(int64_t hashCode, bool bSta
{
if(_activeProxys.empty())
{
TLOGERROR("[TARS][EndpointManager::getHashProxyForWeight _activeEndpoints is empty], bStatic:" << bStatic << endl);
TLOGERROR("[EndpointManager::getHashProxyForWeight _activeEndpoints is empty], bStatic:" << bStatic << endl);
return NULL;
}
@ -1047,7 +1047,7 @@ AdapterProxy* EndpointManager::getConHashProxyForWeight(int64_t hashCode, bool b
{
if(_vRegProxys.empty())
{
TLOGERROR("[TARS][EndpointManager::getConHashProxyForWeight _vRegProxys is empty], bStatic:" << bStatic << endl);
TLOGERROR("[EndpointManager::getConHashProxyForWeight _vRegProxys is empty], bStatic:" << bStatic << endl);
return NULL;
}
@ -1059,7 +1059,7 @@ AdapterProxy* EndpointManager::getConHashProxyForWeight(int64_t hashCode, bool b
int64_t iEnd = TNOWMS;
TLOGTARS("[TARS][EndpointManager::getConHashProxyForWeight update bStatic:" << bStatic << "|_objName:" << _objName << "|timecost(ms):" << (iEnd - iBegin) << endl);
TLOGTARS("[EndpointManager::getConHashProxyForWeight update bStatic:" << bStatic << "|_objName:" << _objName << "|timecost(ms):" << (iEnd - iBegin) << endl);
}
if(_consistentHashWeight.size() > 0)
@ -1083,7 +1083,7 @@ AdapterProxy* EndpointManager::getConHashProxyForWeight(int64_t hashCode, bool b
{
if(_activeProxys.empty())
{
TLOGERROR("[TARS][EndpointManager::getConHashProxyForWeight _activeEndpoints is empty], bStatic:" << bStatic << endl);
TLOGERROR("[EndpointManager::getConHashProxyForWeight _activeEndpoints is empty], bStatic:" << bStatic << endl);
return NULL;
}
@ -1197,7 +1197,7 @@ void EndpointManager::updateHashProxyWeighted(bool bStatic)
{
if(_vRegProxys.size() <= 0)
{
TLOGERROR("[TARS][EndpointManager::updateHashProxyWeighted _vRegProxys is empty], bStatic:" << bStatic << endl);
TLOGERROR("[EndpointManager::updateHashProxyWeighted _vRegProxys is empty], bStatic:" << bStatic << endl);
return ;
}
@ -1220,7 +1220,7 @@ void EndpointManager::updateHashProxyWeighted(bool bStatic)
if(vRegProxys.size() <= 0)
{
TLOGERROR("[TARS][EndpointManager::updateHashProxyWeighted vRegProxys is empty], bStatic:" << bStatic << endl);
TLOGERROR("[EndpointManager::updateHashProxyWeighted vRegProxys is empty], bStatic:" << bStatic << endl);
return ;
}
@ -1285,7 +1285,7 @@ void EndpointManager::updateHashProxyWeighted(bool bStatic)
}
}
TLOGTARS("[TARS]EndpointManager::updateHashProxyWeighted bStatic:" << bStatic << "|_objName:" << _objName << "|endpoint:" << vRegProxys[i]->endpoint().desc() << "|iWeight:" << vRegProxys[i]->getWeight() << "|iWeightR:" << iWeight << "|iIndex:" << vIndex[i] << endl);
TLOGTARS("EndpointManager::updateHashProxyWeighted bStatic:" << bStatic << "|_objName:" << _objName << "|endpoint:" << vRegProxys[i]->endpoint().desc() << "|iWeight:" << vRegProxys[i]->getWeight() << "|iWeightR:" << iWeight << "|iIndex:" << vIndex[i] << endl);
}
for(size_t i = 0; i < iMaxRouterR; i++)
@ -1322,7 +1322,7 @@ void EndpointManager::updateConHashProxyWeighted(bool bStatic, vector<AdapterPro
{
if(_vRegProxys.size() <= 0)
{
TLOGERROR("[TARS][EndpointManager::updateHashProxyWeighted _vRegProxys is empty], bStatic:" << bStatic << endl);
TLOGERROR("[EndpointManager::updateHashProxyWeighted _vRegProxys is empty], bStatic:" << bStatic << endl);
return ;
}
@ -1350,7 +1350,7 @@ AdapterProxy* EndpointManager::getHashProxyForNormal(int64_t hashCode, bool only
{
if(_vRegProxys.empty())
{
TLOGERROR("[TARS][EndpointManager::getHashProxyForNormal _vRegProxys is empty]" << endl);
TLOGERROR("[EndpointManager::getHashProxyForNormal _vRegProxys is empty]" << endl);
return NULL;
}
@ -1374,7 +1374,7 @@ AdapterProxy* EndpointManager::getHashProxyForNormal(int64_t hashCode, bool only
{
if(_activeProxys.empty())
{
TLOGERROR("[TARS][EndpointManager::getHashProxyForNormal _activeEndpoints is empty]" << endl);
TLOGERROR("[EndpointManager::getHashProxyForNormal _activeEndpoints is empty]" << endl);
return NULL;
}
@ -1436,7 +1436,7 @@ AdapterProxy* EndpointManager::getConHashProxyForNormal(int64_t hashCode, bool o
{
if(_vRegProxys.empty())
{
TLOGERROR("[TARS][EndpointManager::getConHashProxyForNormal _vRegProxys is empty]" << endl);
TLOGERROR("[EndpointManager::getConHashProxyForNormal _vRegProxys is empty]" << endl);
return NULL;
}
@ -1448,7 +1448,7 @@ AdapterProxy* EndpointManager::getConHashProxyForNormal(int64_t hashCode, bool o
int64_t iEnd = TNOWMS;
TLOGTARS("[TARS][EndpointManager::getConHashProxyForNormal update _objName:" << _objName << "|timecost(ms):" << (iEnd - iBegin) << endl);
TLOGTARS("[EndpointManager::getConHashProxyForNormal update _objName:" << _objName << "|timecost(ms):" << (iEnd - iBegin) << endl);
}
if(_consistentHash.size() > 0)
@ -1472,7 +1472,7 @@ AdapterProxy* EndpointManager::getConHashProxyForNormal(int64_t hashCode, bool o
{
if(_activeProxys.empty())
{
TLOGERROR("[TARS][EndpointManager::getConHashProxyForNormal _activeEndpoints is empty]" << endl);
TLOGERROR("[EndpointManager::getConHashProxyForNormal _activeEndpoints is empty]" << endl);
return NULL;
}
@ -1543,7 +1543,7 @@ AdapterProxy* EndpointManager::getWeightedForNormal(bool bStaticWeighted, bool o
{
if (_activeProxys.empty())
{
TLOGERROR("[TARS][EndpointManager::getWeightedForNormal activeEndpoints is empty][obj:"<<_objName<<"]" << endl);
TLOGERROR("[EndpointManager::getWeightedForNormal activeEndpoints is empty][obj:"<<_objName<<"]" << endl);
return NULL;
}
@ -1671,7 +1671,7 @@ void EndpointManager::updateProxyWeighted()
if(iWeightProxySize <= 0)
{
TLOGERROR("[TARS]EndpointManager::updateProxyWeighted _objName:" << _objName << "|_activeProxys.size() <= 0" << endl);
TLOGERROR("[EndpointManager::updateProxyWeighted _objName:" << _objName << ", activeProxys.size() <= 0]" << endl);
return ;
}
@ -1689,7 +1689,7 @@ void EndpointManager::updateProxyWeighted()
if(iWeightProxySize <= 0)
{
TLOGERROR("[TARS]EndpointManager::updateProxyWeighted _objName:" << _objName << "|vProxy.size() <= 0" << endl);
TLOGERROR("[EndpointManager::updateProxyWeighted _objName:" << _objName << ", vProxy.size() <= 0]" << endl);
return ;
}
@ -1722,7 +1722,7 @@ void EndpointManager::dispatchEndpointCache(const vector<int> &vWeight)
{
if(vWeight.size() <= 0)
{
TLOGERROR("EndpointManager::dispatchEndpointCache vWeight.size() < 0" << endl);
TLOGERROR("[EndpointManager::dispatchEndpointCache vWeight.size() < 0]" << endl);
return ;
}
@ -1796,7 +1796,7 @@ void EndpointManager::dispatchEndpointCache(const vector<int> &vWeight)
_staticRouterCache.push_back(i);
}
TLOGTARS("[TARS]EndpointManager::dispatchEndpointCache _objName:" << _objName << "|endpoint:" << _activeWeightProxy[i]->endpoint().desc() << "|iWeightR:" << iWeight << endl);
TLOGTARS("EndpointManager::dispatchEndpointCache _objName:" << _objName << "|endpoint:" << _activeWeightProxy[i]->endpoint().desc() << "|iWeightR:" << iWeight << endl);
}
for(size_t i = 0; i < iMaxRouterR; i++)

26
servant/libservant/KeepAliveNodeF.cpp
View File

@ -85,31 +85,31 @@ void KeepAliveNodeFHelper::keepActiving()
}
catch(exception &ex)
{
LOG->error() << "TafNodeFHelper::keepAlive error:" << ex.what() << endl;
TLOGERROR("[KeepAliveNodeFHelper::keepAlive error:" << ex.what() << "]" << endl);
}
catch(...)
{
LOG->error() << "TafNodeFHelper::keepAlive unknown error" << endl;
TLOGERROR("[KeepAliveNodeFHelper::keepAlive unknown error]" << endl);
}
}
void KeepAliveNodeFHelper::reportVersion(const string &version)
{
try
{
try
{
if(_nodePrx)
{
_nodePrx->async_reportVersion(NULL, _si.application, _si.serverName, version);
}
}
catch(exception &ex)
{
TLOGERROR("KeepAliveNodeFHelper::reportVersion error:" << ex.what() << endl);
}
catch(...)
{
TLOGERROR("KeepAliveNodeFHelper::reportVersion unknown error" << endl);
}
}
catch(exception &ex)
{
TLOGERROR("[KeepAliveNodeFHelper::reportVersion error:" << ex.what() << "]" << endl);
}
catch(...)
{
TLOGERROR("[KeepAliveNodeFHelper::reportVersion unknown error" << "]" << endl);
}
}
}

32
servant/libservant/ObjectProxy.cpp
View File

@ -153,7 +153,7 @@ vector<SocketOpt>& ObjectProxy::getSocketOpt()
//
//bool ObjectProxy::invoke_sync(ReqMessage * msg)
//{
// TLOGTARS("[TARS][ObjectProxy::invoke_sync, " << _name << ", begin]" << endl);
// TLOGTARS("[ObjectProxy::invoke_sync, " << _name << ", begin]" << endl);
//
// //选择一个远程服务的Adapter来调用
// AdapterProxy * pAdapterProxy = NULL;
@ -171,7 +171,7 @@ vector<SocketOpt>& ObjectProxy::getSocketOpt()
void ObjectProxy::invoke(ReqMessage * msg)
{
TLOGTARS("[TARS][ObjectProxy::invoke, objname:" << _name << ", begin...]" << endl);
TLOGTARS("[ObjectProxy::invoke, objname:" << _name << ", begin...]" << endl);
//选择一个远程服务的Adapter来调用
AdapterProxy * pAdapterProxy = NULL;
@ -185,14 +185,14 @@ void ObjectProxy::invoke(ReqMessage * msg)
assert(bRet);
//把数据缓存在obj里面
TLOGTARS("[TARS][ObjectProxy::invoke, objname:" << _name << ", select adapter proxy not valid (have not inovoke reg)]" << endl);
TLOGTARS("[ObjectProxy::invoke, objname:" << _name << ", select adapter proxy not valid (have not invoke reg)]" << endl);
return;
}
if(!pAdapterProxy)
{
TLOGERROR("[TARS][ObjectProxy::invoke, objname:"<< _name << ", selectAdapterProxy is null]"<<endl);
TLOGERROR("[ObjectProxy::invoke, objname:"<< _name << ", selectAdapterProxy is null]"<<endl);
msg->response->iRet = TARSADAPTERNULL;
@ -211,7 +211,7 @@ void ObjectProxy::invoke(ReqMessage * msg)
assert(bRet);
//把数据缓存在obj里面
TLOGTARS("[TARS][ObjectProxy::invoke, " << _name << ", select adapter proxy not connected (have not inovoke reg)]" << endl);
TLOGTARS("[ObjectProxy::invoke, " << _name << ", select adapter proxy not connected (have not inovoke reg)]" << endl);
return;
}
@ -223,7 +223,7 @@ void ObjectProxy::onConnect(AdapterProxy *adapterProxy)
{
while(!_reqTimeoutQueue.empty())
{
TLOGTARS("[TARS][ObjectProxy::doInvoke, " << _name << ", pop...]" << endl);
TLOGTARS("[ObjectProxy::onConnect, " << _name << ", queue size:" << _reqTimeoutQueue.size() << ", pop...]" << endl);
ReqMessage * msg = NULL;
_reqTimeoutQueue.pop(msg);
@ -238,7 +238,7 @@ void ObjectProxy::onConnect(AdapterProxy *adapterProxy)
if (!adapterProxy)
{
//这里肯定是请求过主控
TLOGERROR("[TARS][ObjectProxy::doInvoke, " << _name << ", selectAdapterProxy is null]" << endl);
TLOGERROR("[ObjectProxy::onConnect, " << _name << ", selectAdapterProxy is null]" << endl);
msg->response->iRet = TARSADAPTERNULL;
doInvokeException(msg);
return;
@ -258,13 +258,13 @@ void ObjectProxy::onConnect(AdapterProxy *adapterProxy)
void ObjectProxy::onNotifyEndpoints(const set<EndpointInfo> & active,const set<EndpointInfo> & inactive)
{
if(_servantProxy) {
_servantProxy->onNotifyEndpoints(active, inactive);
_servantProxy->onNotifyEndpoints(this->_communicatorEpoll->getCommunicatorEpollId(), active, inactive);
}
}
void ObjectProxy::doInvoke()
{
TLOGTARS("[TARS][ObjectProxy::doInvoke, objname:" << _name << ", begin...]" << endl);
TLOGTARS("[ObjectProxy::doInvoke, objname:" << _name << ", begin...]" << endl);
for(auto it = _reqTimeoutQueue.begin(); it != _reqTimeoutQueue.end(); ++it)
{
@ -278,7 +278,7 @@ void ObjectProxy::doInvoke()
//
// while(!_reqTimeoutQueue.empty())
// {
// TLOGTARS("[TARS][ObjectProxy::doInvoke, " << _name << ", pop...]" << endl);
// TLOGTARS("[ObjectProxy::doInvoke, " << _name << ", pop...]" << endl);
//
// ReqMessage * msg = NULL;
// _reqTimeoutQueue.pop(msg);
@ -292,7 +292,7 @@ void ObjectProxy::doInvoke()
//
// if (!adapterProxy) {
// //这里肯定是请求过主控
// TLOGERROR("[TARS][ObjectProxy::doInvoke, " << _name << ", selectAdapterProxy is null]" << endl);
// TLOGERROR("[ObjectProxy::doInvoke, " << _name << ", selectAdapterProxy is null]" << endl);
// msg->response->iRet = JCEADAPTERNULL;
// doInvokeException(msg);
// return;
@ -306,7 +306,7 @@ void ObjectProxy::doInvoke()
void ObjectProxy::doInvokeException(ReqMessage * msg)
{
// TLOGTARS("[TARS][ObjectProxy::doInvokeException, objname:" << _name << "]" << endl);
// TLOGTARS("[ObjectProxy::doInvokeException, objname:" << _name << "]" << endl);
//单向调用出现异常直接删除请求
if(msg->eType == ReqMessage::ONE_WAY)
@ -352,11 +352,11 @@ void ObjectProxy::doInvokeException(ReqMessage * msg)
}
catch(exception & e)
{
TLOGERROR("[TARS]ObjectProxy::doInvokeException exp:" << e.what() << " ,line:" << __LINE__ << endl);
TLOGERROR("ObjectProxy::doInvokeException exp:" << e.what() << " ,line:" << __LINE__ << endl);
}
catch(...)
{
TLOGERROR("[TARS]ObjectProxy::doInvokeException exp:unknown line:|" << __LINE__ << endl);
TLOGERROR("ObjectProxy::doInvokeException exp:unknown line:|" << __LINE__ << endl);
}
}
else
@ -378,7 +378,7 @@ void ObjectProxy::doInvokeException(ReqMessage * msg)
}
else
{
TLOGERROR("[TARS]ObjectProxy::doInvokeException coro parallel callback error, objname:" << _name << endl);
TLOGERROR("ObjectProxy::doInvokeException coro parallel callback error, objname:" << _name << endl);
delete msg;
}
}
@ -397,7 +397,7 @@ void ObjectProxy::doTimeout()
ReqMessage * reqInfo = NULL;
while(_reqTimeoutQueue.timeout(reqInfo))
{
TLOGERROR("[TARS][ObjectProxy::doTimeout, objname:" << _name << ", queue timeout error]" << endl);
TLOGERROR("[ObjectProxy::doTimeout, objname:" << _name << ", queue timeout error]" << endl);
reqInfo->response->iRet = TARSINVOKETIMEOUT;

8
servant/libservant/RemoteLogger.cpp
View File

@ -88,7 +88,7 @@ void RollWriteT::operator()(ostream &of, const deque<pair<size_t, string> > &ds)
}
catch(exception &ex)
{
TLOGERROR("[TARS] dyeing log write to remote log server error:" << ex.what() << endl);
TLOGERROR("[dyeing log write to remote log server error:" << ex.what() << "]" << endl);
}
}
}
@ -263,7 +263,7 @@ void RemoteTimeWriteT::sync2remote(const vector<string> &v)
}
catch(exception &ex)
{
TLOGERROR("[TARS] write to remote log server error:" << ex.what() << ": buffer size:" << v.size() << endl);
TLOGERROR("[write to remote log server error:" << ex.what() << ": buffer size:" << v.size() << "]"<< endl);
_timeWrite->writeError(v);
if (_timeWrite->_reportFailPtr)
{
@ -280,7 +280,7 @@ void RemoteTimeWriteT::sync2remoteDyeing(const vector<string> &v)
}
catch(exception &ex)
{
TLOGERROR("[TARS] write dyeing log to remote log server error:" << ex.what() << ": buffer size:" << v.size() << endl);
TLOGERROR("[write dyeing log to remote log server error:" << ex.what() << ": buffer size:" << v.size() << "]" << endl);
_timeWrite->writeError(v);
}
}
@ -422,7 +422,7 @@ void TimeWriteT::operator()(ostream &of, const deque<pair<size_t, string> > &buf
}
catch(exception &ex)
{
TLOGERROR("[TARS] dyeing log write to remote log server error:" << ex.what() << endl);
TLOGERROR("[dyeing log write to remote log server error:" << ex.what() << "]" << endl);
}
}
}

20
servant/libservant/RemoteNotify.cpp
View File

@ -44,6 +44,7 @@ void RemoteNotify::report(const string &sResult, bool bSync)
if(_notifyPrx)
{
ReportInfo info;
info.eType = REPORT;
info.sApp = _app;
info.sServer = _serverName;
info.sSet = _setName;
@ -52,23 +53,21 @@ void RemoteNotify::report(const string &sResult, bool bSync)
info.sNodeName = _nodeName;
if(!bSync)
{
//_notifyPrx->async_reportServer(NULL, _app + "." + _serverName, TC_Common::tostr(std::this_thread::get_id()), sResult);
_notifyPrx->async_reportNotifyInfo(NULL, info);
}
else
{
//_notifyPrx->reportServer(_app + "." + _serverName, TC_Common::tostr(std::this_thread::get_id()), sResult);
_notifyPrx->reportNotifyInfo(info);
}
}
}
catch(exception &ex)
{
TLOGERROR("RemoteNotify::report error:" << ex.what() << endl);
TLOGERROR("[RemoteNotify::report error:" << ex.what() << "]" << endl);
}
catch(...)
{
TLOGERROR("RemoteNotify::report unknown error" << endl);
TLOGERROR("[RemoteNotify::report unknown error" << "]" << endl);
}
}
@ -79,7 +78,7 @@ void RemoteNotify::notify(NOTIFYLEVEL level, const string &sMessage)
if(_notifyPrx)
{
ReportInfo info;
// info.eType = 0;
info.eType = NOTIFY;
info.sApp = _app;
info.sServer = _serverName;
info.sSet = _setName;
@ -87,17 +86,16 @@ void RemoteNotify::notify(NOTIFYLEVEL level, const string &sMessage)
info.sMessage = sMessage;
info.eLevel = level;
info.sNodeName = _nodeName;
//_notifyPrx->async_notifyServer(NULL, _app + "." + _serverName, level, sMessage);
_notifyPrx->async_reportNotifyInfo(NULL, info);
}
}
catch(exception &ex)
{
TLOGERROR("RemoteNotify::notify error:" << ex.what() << endl);
TLOGERROR("[RemoteNotify::notify error:" << ex.what() << "]" << endl);
}
catch(...)
{
TLOGERROR("RemoteNotify::notify unknown error" << endl);
TLOGERROR("[RemoteNotify::notify unknown error" << "]" << endl);
}
}
@ -108,7 +106,7 @@ void RemoteNotify::report(const string &sMessage, const string & app, const stri
if(_notifyPrx)
{
ReportInfo info;
// info.eType = 0;
info.eType = REPORT;
info.sApp = app;
info.sServer = serverName;
info.sSet = "";
@ -119,11 +117,11 @@ void RemoteNotify::report(const string &sMessage, const string & app, const stri
}
catch(exception &ex)
{
TLOGERROR("RemoteNotify::notify error:" << ex.what() << endl);
TLOGERROR("[RemoteNotify::notify error:" << ex.what() << "]" << endl);
}
catch(...)
{
TLOGERROR("RemoteNotify::notify unknown error" << endl);
TLOGERROR("[RemoteNotify::notify unknown error" << "]" << endl);
}
}

2
servant/libservant/Servant.cpp
View File

@ -71,7 +71,7 @@ int Servant::dispatch(TarsCurrentPtr current, vector<char> &buffer)
if (current->getFuncName() == "tars_ping")
{
TLOGTARS("[TARS][Servant::dispatch] tars_ping ok from [" << current->getIp() << ":" << current->getPort() << "]" << endl);
TLOGTARS("[Servant::dispatch] tars_ping ok from [" << current->getIp() << ":" << current->getPort() << "]" << endl);
ret = TARSSERVERSUCCESS;
}

131
servant/libservant/ServantHandle.cpp
View File

@ -50,11 +50,11 @@ ServantHandle::~ServantHandle()
}
catch(exception &ex)
{
TLOGERROR("[TARS]ServantHandle::destroy error:" << ex.what() << endl);
TLOGERROR("[ServantHandle::destroy error:" << ex.what() << "]" << endl);
}
catch(...)
{
TLOGERROR("[TARS]ServantHandle::destroy unknown exception error" << endl);
TLOGERROR("[ServantHandle::destroy unknown exception error]" << endl);
}
++it;
}
@ -114,13 +114,13 @@ void ServantHandle::run()
}
catch(exception &ex)
{
TLOGERROR("[TARS]ServantHandle::run exception error:" << ex.what() << endl);
cerr << "[TARS]ServantHandle::run exception error:" << ex.what() << endl;
TLOGERROR("[ServantHandle::run exception error:" << ex.what() << "]" << endl);
cerr << "ServantHandle::run exception error:" << ex.what() << endl;
}
catch(...)
{
TLOGERROR("[TARS]ServantHandle::run unknown exception error." << endl);
cerr << "[TARS]ServantHandle::run unknown exception error." << endl;
TLOGERROR("[ServantHandle::run unknown exception error]" << endl);
cerr << "ServantHandle::run unknown exception error]" << endl;
}
}
@ -241,11 +241,11 @@ void ServantHandle::handleRecvData(const shared_ptr<TC_EpollServer::RecvContext>
}
catch(exception &ex)
{
TLOGERROR("[TARS]ServantHandle::handleRecvData exception:" << ex.what() << endl);
TLOGERROR("[ServantHandle::handleRecvData exception:" << ex.what() << "]" << endl);
}
catch(...)
{
TLOGERROR("[TARS]ServantHandle::handleRecvData unknown exception error" << endl);
TLOGERROR("[ServantHandle::handleRecvData unknown exception error]" << endl);
}
}
@ -276,11 +276,11 @@ void ServantHandle::handleAsyncResponse()
}
catch (exception& e)
{
TLOGERROR("[TARS][ServantHandle::doResponse ex:" << e.what() << "]" << endl);
TLOGERROR("[ServantHandle::doResponse ex:" << e.what() << "]" << endl);
}
catch (...)
{
TLOGERROR("[TARS][ServantHandle::doResponse ex.]" << endl);
TLOGERROR("[ServantHandle::doResponse error]" << endl);
}
}
@ -292,11 +292,11 @@ void ServantHandle::handleAsyncResponse()
}
catch (exception& e)
{
TLOGERROR("[TARS][ServantHandle::doCustemMessage ex:" << e.what() << "]" << endl);
TLOGERROR("[ServantHandle::doCustemMessage ex:" << e.what() << "]" << endl);
}
catch (...)
{
TLOGERROR("[TARS][ServantHandle::doCustemMessage ex.]" << endl);
TLOGERROR("[ServantHandle::doCustemMessage ex.]" << endl);
}
++it;
@ -315,11 +315,11 @@ void ServantHandle::handleCustomMessage(bool bExpectIdle)
}
catch (exception& e)
{
TLOGERROR("[TARS][ServantHandle::doCustemMessage ex:" << e.what() << "]" << endl);
TLOGERROR("[ServantHandle::doCustemMessage ex:" << e.what() << "]" << endl);
}
catch (...)
{
TLOGERROR("[TARS][ServantHandle::doCustemMessage ex.]" << endl);
TLOGERROR("[ServantHandle::doCustemMessage ex.]" << endl);
}
}
}
@ -349,8 +349,8 @@ void ServantHandle::initialize()
}
else
{
TLOGERROR("[TARS]ServantHandle initialize createServant ret null, for adapter `" +_bindAdapter->getName() + "`" << endl);
cerr << "[TARS]ServantHandle initialize createServant ret null, for adapter `" +_bindAdapter->getName() + "`" << endl;
TLOGERROR("[ServantHandle initialize createServant ret null, for adapter `" +_bindAdapter->getName() + "`]" << endl);
cerr << "ServantHandle initialize createServant ret null, for adapter `" +_bindAdapter->getName() + "`]" << endl;
RemoteNotify::getInstance()->report("initialize createServant error: no adapter:" + _bindAdapter->getName());
@ -363,7 +363,7 @@ void ServantHandle::initialize()
if(it == _servants.end())
{
TLOGERROR("[TARS]initialize error: no servant exists." << endl);
TLOGERROR("[initialize error: no servant exists]" << endl);
RemoteNotify::getInstance()->report("initialize error: no servant exists.");
@ -380,11 +380,11 @@ void ServantHandle::initialize()
it->second->initialize();
TLOGTARS("[TARS][" << it->second->getName() << "] initialize" << endl);
TLOGTARS("[" << it->second->getName() << " initialize]" << endl);
}
catch(exception &ex)
{
TLOGERROR("[TARS]initialize error:" << ex.what() << endl);
TLOGERROR("[initialize error:" << ex.what() << "]" << endl);
RemoteNotify::getInstance()->report("initialize error:" + string(ex.what()));
@ -394,7 +394,7 @@ void ServantHandle::initialize()
}
catch(...)
{
TLOGERROR("[TARS]initialize unknown exception error" << endl);
TLOGERROR("[initialize unknown exception error]" << endl);
RemoteNotify::getInstance()->report("initialize unknown exception error");
@ -416,8 +416,6 @@ void ServantHandle::heartbeat()
TARS_KEEPALIVE(_bindAdapter->getName());
// TLOGERROR("[TARS]ServantHandle::handle heartbeat:" << _bindAdapter->getName() << endl);
//上报连接数 比率
if (_bindAdapter->_pReportConRate)
{
@ -442,7 +440,7 @@ TarsCurrentPtr ServantHandle::createCurrent(const shared_ptr<TC_EpollServer::Rec
}
catch (TarsDecodeException &ex)
{
TLOGERROR("[TARS]ServantHandle::handle request protocol decode error:" << ex.what() << endl);
TLOGERROR("[ServantHandle::handle request protocol decode error:" << ex.what() << "]" << endl);
close(data);
return NULL;
}
@ -459,13 +457,13 @@ TarsCurrentPtr ServantHandle::createCurrent(const shared_ptr<TC_EpollServer::Rec
if(data->adapter()->_pReportTimeoutNum)
data->adapter()->_pReportTimeoutNum->report(1);
TLOGERROR("[TARS]ServantHandle::handle queue timeout:"
<< current->_request.sServantName << "|"
<< current->_request.sFuncName << "|"
<< data->recvTimeStamp() << "|"
<< data->adapter()->getQueueTimeout() << "|"
<< current->_request.iTimeout << "|"
<< now << "|" << data->ip() << ":" << data->port() << endl);
TLOGERROR("[ServantHandle::handle queue timeout:"
<< current->_request.sServantName << ", func:"
<< current->_request.sFuncName << ", recv time:"
<< data->recvTimeStamp() << ", queue timeout:"
<< data->adapter()->getQueueTimeout() << ", timeout:"
<< current->_request.iTimeout << ", now:"
<< now << ", ip:" << data->ip() << ", port:" << data->port() << "]" << endl);
current->sendResponse(TARSSERVERQUEUETIMEOUT);
@ -488,8 +486,7 @@ TarsCurrentPtr ServantHandle::createCloseCurrent(const shared_ptr<TC_EpollServer
void ServantHandle::handleClose(const shared_ptr<TC_EpollServer::RecvContext> &data)
{
TLOGTARS("[TARS]ServantHandle::handleClose,adapter:" << data->adapter()->getName()
<< ",peer:" << data->ip() << ":" << data->port() << endl);
TLOGTARS("[ServantHandle::handleClose,adapter:" << data->adapter()->getName() << ",peer:" << data->ip() << ":" << data->port() << "]"<< endl);
TarsCurrentPtr current = createCloseCurrent(data);
@ -497,8 +494,8 @@ void ServantHandle::handleClose(const shared_ptr<TC_EpollServer::RecvContext> &d
if (sit == _servants.end())
{
TLOGERROR("[TARS]ServantHandle::handleClose,adapter:" << data->adapter()->getName()
<< ",peer:" << data->ip() << ":" << data->port()<<", " << current->getServantName() << " not found" << endl);
TLOGERROR("[ServantHandle::handleClose,adapter:" << data->adapter()->getName()
<< ",peer:" << data->ip() << ":" << data->port()<<", " << current->getServantName() << " not found]" << endl);
return;
}
@ -511,13 +508,13 @@ void ServantHandle::handleClose(const shared_ptr<TC_EpollServer::RecvContext> &d
}
catch(exception &ex)
{
TLOGERROR("[TARS]ServantHandle::handleClose " << ex.what() << endl);
TLOGERROR("[ServantHandle::handleClose " << ex.what() << "]" << endl);
return;
}
catch(...)
{
TLOGERROR("[TARS]ServantHandle::handleClose unknown error" << endl);
TLOGERROR("[ServantHandle::handleClose unknown error]" << endl);
return;
}
@ -533,11 +530,11 @@ void ServantHandle::handleTimeout(const shared_ptr<TC_EpollServer::RecvContext>
if(data->adapter()->_pReportTimeoutNum)
data->adapter()->_pReportTimeoutNum->report(1);
TLOGERROR("[TARS]ServantHandle::handleTimeout adapter '"
TLOGERROR("[ServantHandle::handleTimeout adapter '"
<< data->adapter()->getName()
<< "', recvtime:" << data->recvTimeStamp() << "|"
<< ", timeout:" << data->adapter()->getQueueTimeout()
<< ", id:" << current->getRequestId() << endl);
<< ", id:" << current->getRequestId() << "]" << endl);
if (current->getBindAdapter()->isTarsProtocol())
{
@ -551,11 +548,11 @@ void ServantHandle::handleOverload(const shared_ptr<TC_EpollServer::RecvContext>
if (!current) return;
TLOGERROR("[TARS]ServantHandle::handleOverload adapter '"
TLOGERROR("[ServantHandle::handleOverload adapter '"
<< data->adapter()->getName()
<< "',overload:-1,queue capacity:"
<< data->adapter()->getQueueCapacity()
<< ",id:" << current->getRequestId() << endl);
<< ",id:" << current->getRequestId() << "]" << endl);
if (current->getBindAdapter()->isTarsProtocol())
{
@ -675,19 +672,17 @@ bool ServantHandle::processDye(const TarsCurrentPtr &current, string& dyeingKey)
if (IS_MSG_TYPE(current->getMessageType(), tars::TARSMESSAGETYPEDYED))
{
TLOGTARS("[TARS] servant got a dyeing request, message_type set: " << current->getMessageType() << endl);
TLOGTARS("[servant got a dyeing request, message_type set: " << current->getMessageType() << "]" << endl);
if (dyeingIt != current->getRequestStatus().end())
{
TLOGTARS("[TARS] servant got a dyeing request, dyeing key: " << dyeingIt->second << endl);
TLOGTARS("[servant got a dyeing request, dyeing key: " << dyeingIt->second << "]" << endl);
dyeingKey = dyeingIt->second;
}
return true;
}
return false;
}
@ -704,24 +699,6 @@ bool ServantHandle::processCookie(const TarsCurrentPtr &current, map<string, str
});
return !cookie.empty();
//
// map<string, string>::const_iterator cookieIt = current->getRequestStatus().find(ServantProxy::STATUS_COOKIE);
// if (cookieIt != current->getContext().end())
// {
// std::for_each(pSptd->_cookie.begin(), pSptd->_cookie.end(),[&](map<string, string>::value_type& p){
// req.status.insert(make_pair(p.first, p.second));
// });
//
//// TLOGINFO("[TARS] cookie:" << cookieIt->second << endl);
//
// Cookie stCookie;
// stCookie.readFromJsonString(cookieIt->second);
// cookie = stCookie.cookie;
// return true;
// }
//
// return false;
}
bool ServantHandle::checkValidSetInvoke(const TarsCurrentPtr &current)
@ -749,7 +726,7 @@ bool ServantHandle::checkValidSetInvoke(const TarsCurrentPtr &current)
if (setIt != current->getRequestStatus().end())
{
TLOGTARS("[TARS] servant got a setname request, setname key:" << setIt->second << endl);
TLOGTARS("[servant got a setname request, setname key:" << setIt->second << "]" << endl);
sSetName = setIt->second;
@ -774,13 +751,13 @@ bool ServantHandle::checkValidSetInvoke(const TarsCurrentPtr &current)
}
else
{
TLOGERROR("[TARS]ServantHandle::checkValidSetInvoke|"
TLOGERROR("[ServantHandle::checkValidSetInvoke|"
<< current->getIp() << "|"
<< current->getMessageType() << "|"
<< current->getServantName() << "|"
<< current->getFuncName() << "|client:"
<< ClientConfig::SetDivision << "|server:"
<< sSetName << endl);
<< sSetName << "]" << endl);
current->sendResponse(TARSINVOKEBYINVALIDESET);
return false;
}
@ -788,13 +765,13 @@ bool ServantHandle::checkValidSetInvoke(const TarsCurrentPtr &current)
}
else
{
TLOGERROR("[TARS]ServantHandle::checkValidSetInvoke|"
TLOGERROR("[ServantHandle::checkValidSetInvoke|"
<< current->getIp() << "|"
<< current->getMessageType() << "|"
<< current->getServantName() << "|"
<< current->getFuncName() << "|client:"
<< ClientConfig::SetDivision << "|server:"
<< sSetName << endl);
<< sSetName << "]" << endl);
current->sendResponse(TARSINVOKEBYINVALIDESET);
return false;
}
@ -806,14 +783,14 @@ bool ServantHandle::checkValidSetInvoke(const TarsCurrentPtr &current)
void ServantHandle::handleTarsProtocol(const TarsCurrentPtr &current)
{
TLOGTARS("[TARS]ServantHandle::handleTarsProtocol current:"
TLOGTARS("[ServantHandle::handleTarsProtocol current:"
<< current->getIp() << "|"
<< current->getPort() << "|"
<< current->getMessageType() << "|"
<< current->getServantName() << "|"
<< current->getFuncName() << "|"
<< current->getRequestId() << "|"
<< TC_Common::tostr(current->getRequestStatus())<<endl);
<< TC_Common::tostr(current->getRequestStatus()) << "]"<<endl);
//检查set调用合法性
if(!checkValidSetInvoke(current))
@ -865,7 +842,7 @@ void ServantHandle::handleTarsProtocol(const TarsCurrentPtr &current)
}
catch(TarsDecodeException &ex)
{
TLOGERROR("[TARS]ServantHandle::handleTarsProtocol " << ex.what() << endl);
TLOGERROR("[ServantHandle::handleTarsProtocol " << ex.what() << "]" << endl);
ret = TARSSERVERDECODEERR;
@ -873,7 +850,7 @@ void ServantHandle::handleTarsProtocol(const TarsCurrentPtr &current)
}
catch(TarsEncodeException &ex)
{
TLOGERROR("[TARS]ServantHandle::handleTarsProtocol " << ex.what() << endl);
TLOGERROR("[ServantHandle::handleTarsProtocol " << ex.what() << "]" << endl);
ret = TARSSERVERENCODEERR;
@ -881,7 +858,7 @@ void ServantHandle::handleTarsProtocol(const TarsCurrentPtr &current)
}
catch(exception &ex)
{
TLOGERROR("[TARS]ServantHandle::handleTarsProtocol " << ex.what() << endl);
TLOGERROR("[ServantHandle::handleTarsProtocol " << ex.what() << "]" << endl);
ret = TARSSERVERUNKNOWNERR;
@ -889,7 +866,7 @@ void ServantHandle::handleTarsProtocol(const TarsCurrentPtr &current)
}
catch(...)
{
TLOGERROR("[TARS]ServantHandle::handleTarsProtocol unknown error" << endl);
TLOGERROR("[ServantHandle::handleTarsProtocol unknown error]" << endl);
ret = TARSSERVERUNKNOWNERR;
@ -908,10 +885,10 @@ void ServantHandle::handleTarsProtocol(const TarsCurrentPtr &current)
void ServantHandle::handleNoTarsProtocol(const TarsCurrentPtr &current)
{
TLOGTARS("[TARS]ServantHandle::handleNoTarsProtocol current:"
TLOGTARS("[ServantHandle::handleNoTarsProtocol current:"
<< current->getIp() << "|"
<< current->getPort() << "|"
<< current->getServantName() << endl);
<< current->getServantName() << "]" << endl);
auto sit = _servants.find(current->getServantName());
@ -926,11 +903,11 @@ void ServantHandle::handleNoTarsProtocol(const TarsCurrentPtr &current)
}
catch(exception &ex)
{
TLOGERROR("[TARS]ServantHandle::handleNoTarsProtocol " << ex.what() << endl);
TLOGERROR("[ServantHandle::handleNoTarsProtocol " << ex.what() << "]" << endl);
}
catch(...)
{
TLOGERROR("[TARS]ServantHandle::handleNoTarsProtocol unknown error" << endl);
TLOGERROR("[ServantHandle::handleNoTarsProtocol unknown error]" << endl);
}
if (current->isResponse() && !buffer.empty())

101
servant/libservant/ServantProxy.cpp
View File

@ -210,7 +210,7 @@ void coroWhenAll(const CoroParallelBasePtr &ptr)
{
if(!ptr->checkAllReqSend())
{
TLOGERROR("[TARS][coroWhenAll use coro invoke interface's num not equal ptr set value]"<<endl);
TLOGERROR("[coroWhenAll use coro invoke interface's num not equal ptr set value]"<<endl);
throw TarsUseCoroException("use coro invoke interface's num not equal ptr set value");
}
@ -220,7 +220,7 @@ void coroWhenAll(const CoroParallelBasePtr &ptr)
if(!pSptd->_sched)
{
TLOGERROR("[TARS][coroWhenAll no open coroutine mode]"<<endl);
TLOGERROR("[coroWhenAll no open coroutine mode]"<<endl);
throw TarsUseCoroException("coroWhenAll not open coroutine mode");
}
@ -598,7 +598,7 @@ void ServantProxy::tars_set_push_callback(const ServantProxyCallbackPtr & cb)
// }
// else
// {
// TLOGERROR("[TARS][ServantProxy::invoke_async use coroutine's callback not set CoroParallelBasePtr]" << endl);
// TLOGERROR("[ServantProxy::invoke_async use coroutine's callback not set CoroParallelBasePtr]" << endl);
// delete msg;
// msg = NULL;
// throw TarsUseCoroException("ServantProxy::invoke_async use coroutine's callback not set CoroParallelBasePtr");
@ -606,7 +606,7 @@ void ServantProxy::tars_set_push_callback(const ServantProxyCallbackPtr & cb)
// }
// else
// {
// TLOGERROR("[TARS][ServantProxy::invoke coroutine mode invoke not open]" << endl);
// TLOGERROR("[ServantProxy::invoke coroutine mode invoke not open]" << endl);
// delete msg;
// msg = NULL;
// throw TarsUseCoroException("coroutine mode invoke not open");
@ -617,7 +617,7 @@ void ServantProxy::tars_set_push_callback(const ServantProxyCallbackPtr & cb)
// bool bEmpty;
// if (!pReqQ->push_back(msg, bEmpty))
// {
// TLOGERROR("[TARS][ServantProxy::invoke_async msgQueue push_back error num:" << pSptd->_netSeq << "]" << endl);
// TLOGERROR("[ServantProxy::invoke_async msgQueue push_back error num:" << pSptd->_netSeq << "]" << endl);
// msg->pObjectProxy->getCommunicatorEpoll()->notify(pSptd->_reqQNo, pReqQ);
// delete msg;
// throw TarsClientQueueException("client queue full");
@ -650,7 +650,7 @@ void ServantProxy::invoke(ReqMessage * msg, bool bCoroAsync)
if(msg->bDyeing)
{
TLOGTARS("[TARS][ServantProxy::invoke, set dyeing, key=" << pSptd->_dyeingKey << endl);
TLOGTARS("[ServantProxy::invoke, set dyeing, key=" << pSptd->_dyeingKey << endl);
}
msg->cookie = pSptd->_cookie;
@ -685,7 +685,7 @@ void ServantProxy::invoke(ReqMessage * msg, bool bCoroAsync)
SET_MSG_TYPE(msg->request.iMessageType, TARSMESSAGETYPESETNAME);
msg->request.status[ServantProxy::STATUS_SETNAME_VALUE] = pObjProxy->getInvokeSetName();
TLOGTARS("[TARS][ServantProxy::invoke, " << msg->request.sServantName << ", invoke with set,"<<pObjProxy->getInvokeSetName()<<"]" << endl);
TLOGTARS("[ServantProxy::invoke, " << msg->request.sServantName << ", invoke with set,"<<pObjProxy->getInvokeSetName()<<"]" << endl);
}
//同步调用 new 一个ReqMonitor
@ -724,7 +724,7 @@ void ServantProxy::invoke(ReqMessage * msg, bool bCoroAsync)
}
else
{
TLOGERROR("[TARS][ServantProxy::invoke use coroutine's callback not set CoroParallelBasePtr]"<<endl);
TLOGERROR("[ServantProxy::invoke use coroutine's callback not set CoroParallelBasePtr]"<<endl);
delete msg;
msg = NULL;
throw TarsUseCoroException("use coroutine's callback not set CoroParallelBasePtr");
@ -732,7 +732,7 @@ void ServantProxy::invoke(ReqMessage * msg, bool bCoroAsync)
}
else
{
TLOGERROR("[TARS][ServantProxy::invoke coroutine mode invoke not open]"<<endl);
TLOGERROR("[ServantProxy::invoke coroutine mode invoke not open]"<<endl);
delete msg;
msg = NULL;
throw TarsUseCoroException("coroutine mode invoke not open");
@ -746,7 +746,7 @@ void ServantProxy::invoke(ReqMessage * msg, bool bCoroAsync)
if(!pReqQ->push_back(msg,bEmpty))
{
TLOGERROR("[TARS][ServantProxy::invoke msgQueue push_back error num:" << pSptd->_netSeq << "]" << endl);
TLOGERROR("[ServantProxy::invoke msgQueue push_back error num:" << pSptd->_netSeq << "]" << endl);
delete msg;
msg = NULL;
@ -783,7 +783,7 @@ void ServantProxy::invoke(ReqMessage * msg, bool bCoroAsync)
//判断eStatus来判断状态
assert(msg->eStatus != ReqMessage::REQ_REQ);
TLOGTARS("[TARS]ServantProxy::invoke line: " << __LINE__ << " status: " << msg->eStatus << ", ret: " <<msg->response->iRet << endl);
// TLOGTARS("[ServantProxy::invoke line: " << __LINE__ << " status: " << msg->eStatus << ", ret: " <<msg->response->iRet << endl);
if(msg->eStatus == ReqMessage::REQ_RSP && msg->response->iRet == TARSSERVERSUCCESS)
{
@ -960,55 +960,52 @@ void ServantProxy::rpc_call_async(uint32_t iRequestId,
ServantPrx ServantProxy::getServantPrx(ReqMessage *msg)
{
if(_connectionSerial <= 0)
return this;
if(_servantId == 0)
{
std::lock_guard<std::mutex> m(_servantMutex);
if(_servantId == 0)
{
if(_servantList.empty())
{
for(int i = 0; i < _connectionSerial; ++i)
{
string obj = tars_name() + "#" + TC_Common::tostr(i);
if (!(*_objectProxy)->address().empty())
{
obj += "@" + (*_objectProxy)->address();
}
if(_servantId == 0 && _servantList.empty())
{
for(int i = 0; i < _connectionSerial; ++i)
{
string obj = tars_name() + "#" + TC_Common::tostr(i);
if (!(*_objectProxy)->address().empty())
{
obj += "@" + (*_objectProxy)->address();
}
ServantPrx prx = _communicator->stringToProxy<ServantPrx>(obj);
prx->tars_set_protocol(tars_get_protocol());
prx->_rootPrx = this;
ServantPrx prx = _communicator->stringToProxy<ServantPrx>(obj);
prx->tars_set_protocol(tars_get_protocol());
prx->_rootPrx = this;
_servantList.push_back(prx);
}
}
++_servantId;
}
_servantList.push_back(prx);
}
}
}
return _servantList[(_servantId++) % _servantList.size()];
// assert((int)_servantList.size() == _connectionSerial);
int id = _servantId % (_servantList.size() + 1);
++_servantId;
if(id == 0)
{
return this;
}
return _servantList[(id-1)];
// return _servantList[(_servantId++) % _servantList.size()];
}
void ServantProxy::onNotifyEndpoints(const set<EndpointInfo> & active,const set<EndpointInfo> & inactive)
void ServantProxy::onNotifyEndpoints(size_t netThreadSeq, const set<EndpointInfo> & active,const set<EndpointInfo> & inactive)
{
if(_rootPrx)
{
for (size_t i = 0; i < _rootPrx->_servantList.size(); i++)
{
_rootPrx->_servantList[i]->onNotifyEndpoints(active, inactive);
}
}
else
{
for (size_t i = 0; i < _objectProxyNum; ++i)
{
_objectProxy[i]->getEndpointManager()->updateEndpoints(active, inactive);
}
}
for (size_t i = 0; i < _servantList.size(); i++)
{
_servantList[i]->_objectProxy[netThreadSeq]->getEndpointManager()->updateEndpoints(active, inactive);
}
// _objectProxy[netThreadSeq]->getEndpointManager()->updateEndpoints(active, inactive);
}
void ServantProxy::onSetInactive(const EndpointInfo& ep)
@ -1189,12 +1186,6 @@ void ServantProxy::checkCookie(RequestPacket& req)
std::for_each(pSptd->_cookie.begin(), pSptd->_cookie.end(),[&](map<string, string>::value_type& p){
req.status.insert(make_pair(p.first, p.second));
});
// Cookie stCookie;
// stCookie.cookie = pSptd->_cookie;
//
// req.status.insert(make_pair());
// req.context[ServantProxy::STATUS_COOKIE] = stCookie.writeToJsonString();
}

28
servant/libservant/StatReport.cpp
View File

@ -82,7 +82,7 @@ void StatReport::report(size_t iSeq,MapStatMicMsg * pmStatMicMsg)
delete pmStatMicMsg;
pmStatMicMsg = NULL;
TLOGERROR("[TARS][StatReport::report] queue full." << endl);
TLOGERROR("[StatReport::report] queue full]" << endl);
}
}
@ -466,7 +466,7 @@ int StatReport::reportMicMsg(MapStatMicMsg& msg,bool bFromClient)
msg.swap(mStatMsg);
}
TLOGTARS("[TARS][StatReport::reportMicMsg get size:" << mStatMsg.size()<<"]"<< endl);
TLOGTARS("[StatReport::reportMicMsg get size:" << mStatMsg.size()<<"]"<< endl);
for(MapStatMicMsg::iterator it = mStatMsg.begin(); it != mStatMsg.end(); it++)
{
const StatMicMsgHead &head = it->first;
@ -477,7 +477,7 @@ int StatReport::reportMicMsg(MapStatMicMsg& msg,bool bFromClient)
{
if(_statPrx)
{
TLOGTARS("[TARS][StatReport::reportMicMsg send size:" << mTemp.size()<<"]"<< endl);
TLOGTARS("[StatReport::reportMicMsg send size:" << mTemp.size()<<"]"<< endl);
_statPrx->tars_set_timeout(_reportTimeout)->async_reportMicMsg(NULL,mTemp,bFromClient, ServerConfig::Context);
}
iLen = iTemLen;
@ -492,14 +492,14 @@ int StatReport::reportMicMsg(MapStatMicMsg& msg,bool bFromClient)
head.displaySimple(os);
os << " ";
mTemp[head].displaySimple(os);
TLOGTARS("[TARS][StatReport::reportMicMsg display:" << os.str() << endl);
TLOGTARS("[StatReport::reportMicMsg display:" << os.str() << "]" << endl);
}
}
if(0 != (int)mTemp.size())
{
if(_statPrx)
{
TLOGTARS("[TARS][StatReport::reportMicMsg send size:" << mTemp.size()<<"]"<< endl);
TLOGTARS("[StatReport::reportMicMsg send size:" << mTemp.size()<<"]"<< endl);
_statPrx->tars_set_timeout(_reportTimeout)->async_reportMicMsg(NULL,mTemp,bFromClient, ServerConfig::Context);
}
}
@ -614,12 +614,12 @@ int StatReport::reportPropMsg()
head.displaySimple(os);
os << " ";
mStatMsg[head].displaySimple(os);
TLOGTARS("[TARS][StatReport::reportPropMsg display:" << os.str() << endl);
TLOGTARS("[StatReport::reportPropMsg display:" << os.str() << "]" << endl);
}
}
}
TLOGTARS("[TARS][StatReport::reportPropMsg get size:" << mStatMsg.size()<<"]"<< endl);
TLOGTARS("[StatReport::reportPropMsg get size:" << mStatMsg.size()<<"]"<< endl);
int iLen = 0;
MapStatPropMsg mTemp;
for(MapStatPropMsg::iterator it = mStatMsg.begin(); it != mStatMsg.end(); it++)
@ -638,7 +638,7 @@ int StatReport::reportPropMsg()
{
if(_propertyPrx)
{
TLOGTARS("[TARS][StatReport::reportPropMsg send size:" << mTemp.size()<<"]"<< endl);
TLOGTARS("[StatReport::reportPropMsg send size:" << mTemp.size()<<"]"<< endl);
_propertyPrx->tars_set_timeout(_reportTimeout)->async_reportPropMsg(NULL,mTemp);
}
iLen = iTemLen;
@ -650,7 +650,7 @@ int StatReport::reportPropMsg()
{
if(_propertyPrx)
{
TLOGTARS("[TARS][StatReport::reportPropMsg send size:" << mTemp.size()<< "]"<< endl);
TLOGTARS("[StatReport::reportPropMsg send size:" << mTemp.size()<< "]"<< endl);
_propertyPrx->tars_set_timeout(_reportTimeout)->async_reportPropMsg(NULL,mTemp);
}
}
@ -677,7 +677,7 @@ int StatReport::reportSampleMsg()
_statSampleMsg.swap(mmStatSampleMsg);
}
TLOGTARS("[TARS][StatReport::reportSampleMsg get size:" << mmStatSampleMsg.size()<<"]"<< endl);
TLOGTARS("[StatReport::reportSampleMsg get size:" << mmStatSampleMsg.size()<<"]"<< endl);
int iLen = 0;
vector<StatSampleMsg> vTemp;
@ -690,7 +690,7 @@ int StatReport::reportSampleMsg()
{
if(_statPrx)
{
TLOGTARS("[TARS][StatReport::reportSampleMsg send size:" << vTemp.size()<< "]"<< endl);
TLOGTARS("[StatReport::reportSampleMsg send size:" << vTemp.size()<< "]"<< endl);
_statPrx->tars_set_timeout(_reportTimeout)->async_reportSampleMsg(NULL,vTemp, ServerConfig::Context);
}
iLen = iTemLen;
@ -702,7 +702,7 @@ int StatReport::reportSampleMsg()
{
if(_statPrx)
{
TLOGTARS("[TARS][StatReport::reportSampleMsg send size:" << vTemp.size()<< "]"<< endl);
TLOGTARS("[StatReport::reportSampleMsg send size:" << vTemp.size()<< "]"<< endl);
_statPrx->tars_set_timeout(_reportTimeout)->async_reportSampleMsg(NULL,vTemp, ServerConfig::Context);
}
}
@ -711,11 +711,11 @@ int StatReport::reportSampleMsg()
}
catch ( exception& e )
{
TLOGERROR("StatReport::reportSampleMsg catch exception:" << e.what() << endl);
TLOGERROR("[StatReport::reportSampleMsg catch exception:" << e.what() << "]" << endl);
}
catch ( ... )
{
TLOGERROR("StatReport::reportSampleMsg catch unkown exception" << endl);
TLOGERROR("[StatReport::reportSampleMsg catch unkown exception]" << endl);
}
return -1;
}

106
servant/libservant/Transceiver.cpp
View File

@ -57,7 +57,7 @@ void Transceiver::checkTimeout()
if(eConnecting == _connStatus && TNOWMS > _conTimeoutTime)
{
//链接超时
TLOGERROR("[TARS][Transceiver::checkTimeout ep:"<<_adapterProxy->endpoint().desc()<<" , connect timeout]"<<endl);
TLOGERROR("[Transceiver::checkTimeout ep:"<<_adapterProxy->endpoint().desc()<<" , connect timeout]"<<endl);
_adapterProxy->setConTimeout(true);
close();
}
@ -118,7 +118,7 @@ void Transceiver::connect()
_fd = fd;
TLOGTARS("[TARS][Transceiver::connect obj:" << _adapterProxy->getObjProxy()->name()
TLOGTARS("[Transceiver::connect obj:" << _adapterProxy->getObjProxy()->name()
<< ",connect:" << _ep.desc() << ",fd:" << _fd << "]" << endl);
// //设置网络qos的dscp标志
@ -134,7 +134,7 @@ void Transceiver::connect()
{
if(setsockopt(_fd,socketOpts[i].level,socketOpts[i].optname, (const char*)socketOpts[i].optval,socketOpts[i].optlen) == -1)
{
TLOGERROR("[TARS][setsockopt error:" << NetworkUtil::errorToString(errno)
TLOGERROR("[setsockopt error:" << NetworkUtil::errorToString(errno)
<< ",objname:" << _adapterProxy->getObjProxy()->name()
<< ",desc:" << _ep.desc()
<< ",fd:" << _fd
@ -152,7 +152,7 @@ void Transceiver::setConnected()
_adapterProxy->setConTimeout(false);
_adapterProxy->addConnExc(false);
TLOGTARS("[TARS][tcp setConnected, " << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << "]" << endl);
TLOGTARS("[tcp setConnected, " << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << "]" << endl);
onConnect();
@ -172,7 +172,7 @@ void Transceiver::onConnect()
if (!_openssl)
{
ObjectProxy* obj = _adapterProxy->getObjProxy();
TLOGERROR("[TARS][onConnect:" << obj->name() << " can't find client SSL_CTX " << endl);
TLOGERROR("[onConnect:" << obj->name() << " can't find client SSL_CTX " << endl);
this->close();
return;
}
@ -186,7 +186,7 @@ void Transceiver::onConnect()
int ret = _openssl->doHandshake(_sendBuffer);
if (ret != 0)
{
TLOGERROR("[TARS] SSL_connect failed " << endl);
TLOGERROR(" SSL_connect failed " << endl);
this->close();
return;
}
@ -194,7 +194,7 @@ void Transceiver::onConnect()
// send the encrypt data from write buffer
if (!_sendBuffer.empty())
{
TLOGTARS("[TARS][Transceiver::onConnect handshake:" << _openssl->isHandshaked() << ", send handshake len:" << _sendBuffer.getBufferLength() << endl);
TLOGTARS("[Transceiver::onConnect handshake:" << _openssl->isHandshaked() << ", send handshake len:" << _sendBuffer.getBufferLength() << endl);
doRequest();
}
@ -209,7 +209,7 @@ void Transceiver::doAuthReq()
{
ObjectProxy* obj = _adapterProxy->getObjProxy();
TLOGTARS("[TARS][Transceiver::doAuthReq obj:" << obj->name() << ", auth type:" << etos((AUTH_TYPE)_adapterProxy->endpoint().authType()) << endl);
TLOGTARS("[Transceiver::doAuthReq obj:" << obj->name() << ", auth type:" << etos((AUTH_TYPE)_adapterProxy->endpoint().authType()) << endl);
if (_adapterProxy->endpoint().authType() == AUTH_TYPENONE)
{
@ -286,7 +286,7 @@ void Transceiver::doAuthReq()
//
// if (ret == TC_NetWorkBuffer::PACKET_ERR)
// {
// TLOGERROR("[TARS][tcp doResponse," << _pAdapterProxy->getObjProxy()->name() << ",fd:" << _iFd << "," << _ep.desc() << ",tcp recv decode error" << endl);
// TLOGERROR("[tcp doResponse," << _pAdapterProxy->getObjProxy()->name() << ",fd:" << _iFd << "," << _ep.desc() << ",tcp recv decode error" << endl);
// msg->eStatus = ReqMessage::REQ_NET;
// msg->response->sResultDesc = "recv packet decode failed";
//
@ -300,7 +300,7 @@ void Transceiver::doAuthReq()
// }
// catch (exception & ex) {
// TLOGERROR(
// "[TARS][tcp doResponse," << _pAdapterProxy->getObjProxy()->name() << ",fd:" << _iFd << "," << _ep.desc() << ",tcp recv decode error:" << ex.what() << endl);
// "[tcp doResponse," << _pAdapterProxy->getObjProxy()->name() << ",fd:" << _iFd << "," << _ep.desc() << ",tcp recv decode error:" << ex.what() << endl);
// msg->eStatus = ReqMessage::REQ_NET;
// msg->response->sResultDesc = "recv packet decode failed";
//
@ -308,7 +308,7 @@ void Transceiver::doAuthReq()
// }
// catch (...) {
// TLOGERROR(
// "[TARS][tcp doResponse," << _pAdapterProxy->getObjProxy()->name() << ",fd:" << _iFd << "," << _ep.desc() << ",tcp recv decode error." << endl);
// "[tcp doResponse," << _pAdapterProxy->getObjProxy()->name() << ",fd:" << _iFd << "," << _ep.desc() << ",tcp recv decode error." << endl);
// msg->eStatus = ReqMessage::REQ_NET;
// msg->response->sResultDesc = "recv packet decode failed";
// }
@ -332,7 +332,7 @@ void Transceiver::finishInvoke(shared_ptr<ResponsePacket> &rsp)
tars::stoe(ret, tmp);
tars::AUTH_STATE newstate = tmp;
TLOGTARS("[TARS]Transceiver::finishInvoke state: " << etos(_authState) << " -> " << etos(newstate) << endl);
TLOGTARS("Transceiver::finishInvoke state: " << etos(_authState) << " -> " << etos(newstate) << endl);
setAuthState(newstate);
if (newstate == AUTH_SUCC)
@ -342,7 +342,7 @@ void Transceiver::finishInvoke(shared_ptr<ResponsePacket> &rsp)
}
else
{
TLOGERROR("[TARS]Transceiver::finishInvoke newstate: " << etos(newstate) << ", error close!\n");
TLOGERROR("Transceiver::finishInvoke newstate: " << etos(newstate) << ", error close!\n");
close();
}
@ -377,7 +377,7 @@ bool Transceiver::sendAuthData(const BasicAuthInfo& info)
int ret = _openssl->write(buff.data(), (uint32_t) buff.size(), _sendBuffer);
if(ret != 0)
{
TLOGERROR("[TARS][Transceiver::sendAuthData ssl write failed, obj:" << _adapterProxy->getObjProxy()->name() << ", error:" << _openssl->getErrMsg() << endl);
TLOGERROR("[Transceiver::sendAuthData ssl write failed, obj:" << _adapterProxy->getObjProxy()->name() << ", error:" << _openssl->getErrMsg() << endl);
return false;
}
}
@ -390,12 +390,12 @@ bool Transceiver::sendAuthData(const BasicAuthInfo& info)
#endif
TLOGTARS("[TARS][sendAuthData:" << objPrx->name() << " len: " << _sendBuffer.getBufferLength() << endl);
TLOGTARS("[sendAuthData:" << objPrx->name() << " len: " << _sendBuffer.getBufferLength() << endl);
int ret = doRequest();
if (ret != 0)
{
TLOGERROR("[TARS][Transceiver::setConnected failed sendRequest for Auth\n");
TLOGERROR("[Transceiver::setConnected failed sendRequest for Auth\n");
close();
return false;
}
@ -439,11 +439,11 @@ void Transceiver::close()
if(second > 0) {
_adapterProxy->getObjProxy()->getCommunicatorEpoll()->reConnect(TNOWMS + second * 1000, this);
TLOGERROR("[TARS][trans close:" << _adapterProxy->getObjProxy()->name() << "," << _ep.desc() << ", reconnect:" << second << "]" << endl);
TLOGERROR("[trans close:" << _adapterProxy->getObjProxy()->name() << "," << _ep.desc() << ", reconnect:" << second << "]" << endl);
}
// else
// {
// TLOGERROR("[TARS][trans close:" << _adapterProxy->getObjProxy()->name() << "," << _ep.desc() << "]" << endl);
// TLOGERROR("[trans close:" << _adapterProxy->getObjProxy()->name() << "," << _ep.desc() << "]" << endl);
// }
}
@ -496,7 +496,7 @@ int Transceiver::sendRequest(const shared_ptr<TC_NetWorkBuffer::Buffer> &buff)
if (isSSL() && !_openssl)
return eRetError;
#endif
TLOGTARS("[TARS][Transceiver::sendRequest failed, obj:" << _adapterProxy->getObjProxy()->name() << ", need auth." << endl);
TLOGTARS("[Transceiver::sendRequest failed, obj:" << _adapterProxy->getObjProxy()->name() << ", need auth." << endl);
return eRetError; // 需要鉴权但还没通过，不能发送非认证消息
}
@ -511,14 +511,14 @@ int Transceiver::sendRequest(const shared_ptr<TC_NetWorkBuffer::Buffer> &buff)
if (isSSL())
{
if(!_openssl->isHandshaked()) {
TLOGTARS("[TARS][Transceiver::sendRequest failed, obj:" << _adapterProxy->getObjProxy()->name() << ", ssl need handshake." << endl);
TLOGTARS("[Transceiver::sendRequest failed, obj:" << _adapterProxy->getObjProxy()->name() << ", ssl need handshake." << endl);
return eRetError;
}
int ret = _openssl->write(buff->buffer(), (uint32_t) buff->length(), _sendBuffer);
if(ret != 0)
{
TLOGERROR("[TARS][Transceiver::sendRequest ssl write failed, obj:" << _adapterProxy->getObjProxy()->name() << ", error:" << _openssl->getErrMsg() << endl);
TLOGERROR("[Transceiver::sendRequest ssl write failed, obj:" << _adapterProxy->getObjProxy()->name() << ", error:" << _openssl->getErrMsg() << endl);
return eRetError;
}
@ -551,7 +551,7 @@ int Transceiver::sendRequest(const shared_ptr<TC_NetWorkBuffer::Buffer> &buff)
}
#endif
// TLOGTARS("[TARS][Transceiver::sendRequest handshake:" << _openssl->isHandshaked() << ", origin length:" << buff->length() << endl);
// TLOGTARS("[Transceiver::sendRequest handshake:" << _openssl->isHandshaked() << ", origin length:" << buff->length() << endl);
int iRet = this->send(buff->buffer(), (uint32_t)buff->length(), 0);
@ -600,13 +600,13 @@ int TcpTransceiver::doResponse()
int ret = _openssl->read(buff, iRet, _sendBuffer);
if (ret != 0)
{
TLOGERROR("[TARS][Transceiver::doResponse SSL_read handshake failed: " << _adapterProxy->getObjProxy()->name() << ", info:" << _openssl->getErrMsg() << endl);
TLOGERROR("[Transceiver::doResponse SSL_read handshake failed: " << _adapterProxy->getObjProxy()->name() << ", info:" << _openssl->getErrMsg() << endl);
close();
return -1;
}
else if(!_sendBuffer.empty())
{
TLOGTARS("[TARS][Transceiver::doResponse SSL_read prehandshake:" << preHandshake << ", handshake:" << _openssl->isHandshaked() << ", send handshake:" << _sendBuffer.getBufferLength() << endl);
TLOGTARS("[Transceiver::doResponse SSL_read prehandshake:" << preHandshake << ", handshake:" << _openssl->isHandshaked() << ", send handshake:" << _sendBuffer.getBufferLength() << endl);
doRequest();
}
@ -640,7 +640,7 @@ int TcpTransceiver::doResponse()
ret = _adapterProxy->getObjProxy()->getProxyProtocol().responseFunc(*rbuf, *rsp.get());
if (ret == TC_NetWorkBuffer::PACKET_ERR) {
TLOGERROR( "[TARS][tcp doResponse," << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << "," << _ep.desc() << ",tcp recv decode error" << endl);
TLOGERROR( "[tcp doResponse," << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << "," << _ep.desc() << ",tcp recv decode error" << endl);
close();
break;
}
@ -667,13 +667,13 @@ int TcpTransceiver::doResponse()
}
}
catch (exception & ex) {
TLOGERROR("[TARS][tcp doResponse," << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << ","
TLOGERROR("[tcp doResponse," << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << ","
<< _ep.desc() << ",tcp recv decode error:" << ex.what() << endl);
close();
}
catch (...) {
TLOGERROR("[TARS][tcp doResponse," << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << ","
TLOGERROR("[tcp doResponse," << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << ","
<< _ep.desc() << ",tcp recv decode error." << endl);
close();
@ -682,7 +682,7 @@ int TcpTransceiver::doResponse()
}
while (iRet>0);
// TLOGTARS("[TARS][tcp doResponse, " << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << ", all recvbuf:" << _recvBuffer.getBufferLength() << "]" << endl);
// TLOGTARS("[tcp doResponse, " << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << ", all recvbuf:" << _recvBuffer.getBufferLength() << "]" << endl);
return 0;
}
@ -725,7 +725,7 @@ int TcpTransceiver::doResponse()
// }
// while (iRet>0);
// TLOGTARS("[TARS][tcp doResponse objname:" << _adapterProxy->getObjProxy()->name()
// TLOGTARS("[tcp doResponse objname:" << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << ",recvbuf:" << _recvBuffer.ReadableSize() << "]" << endl);
// if(!_recvBuffer.IsEmpty())
@ -741,7 +741,7 @@ int TcpTransceiver::doResponse()
// std::string out;
// if (!_openssl->Read(_recvBuffer.ReadAddr(), _recvBuffer.ReadableSize(), out))
// {
// TLOGERROR("[TARS][SSL_connect Failed: " << _adapterProxy->getObjProxy()->name() << endl);
// TLOGERROR("[SSL_connect Failed: " << _adapterProxy->getObjProxy()->name() << endl);
// this->close();
// return -1;
// }
@ -795,14 +795,14 @@ int TcpTransceiver::doResponse()
// }
// catch (exception &ex)
// {
// TLOGERROR("[TARS][tcp doResponse objname:" << _adapterProxy->getObjProxy()->name()
// TLOGERROR("[tcp doResponse objname:" << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << ",desc:" << _ep.desc() << ",tcp recv decode error:" << ex.what() << endl);
// close();
// }
// catch (...)
// {
// TLOGERROR("[TARS][tcp doResponse objname:" << _adapterProxy->getObjProxy()->name()
// TLOGERROR("[tcp doResponse objname:" << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << ",desc" << _ep.desc() << ",tcp recv decode error." << endl);
// close();
@ -823,7 +823,7 @@ int TcpTransceiver::send(const void* buf, uint32_t len, uint32_t flag)
if (iRet < 0 && !TC_Socket::isPending())
{
TLOGTARS("[TARS][tcp send," << _adapterProxy->getObjProxy()->name() << ", fd:" << _fd << "," << _ep.desc()
TLOGTARS("[tcp send," << _adapterProxy->getObjProxy()->name() << ", fd:" << _fd << "," << _ep.desc()
<< ", fail! errno:" << TC_Exception::getSystemCode() << ", "
<< TC_Exception::parseError(TC_Exception::getSystemCode()) << ", close]" << endl);
@ -838,7 +838,7 @@ int TcpTransceiver::send(const void* buf, uint32_t len, uint32_t flag)
_adapterProxy->getObjProxy()->getCommunicatorEpoll()->modFd(_fd, &_fdInfo, EPOLLIN | EPOLLOUT);
}
#endif
TLOGTARS("[TARS][tcp send," << _adapterProxy->getObjProxy()->name() << ", fd:" << _fd << ","
TLOGTARS("[tcp send," << _adapterProxy->getObjProxy()->name() << ", fd:" << _fd << ","
<< _ep.desc() << ", len:" << iRet << "]" << endl);
return iRet;
@ -854,7 +854,7 @@ int TcpTransceiver::send(const void* buf, uint32_t len, uint32_t flag)
// if (iRet == 0 || (iRet < 0 && errno != EAGAIN))
// {
// TLOGTARS("[TARS][tcp readv, " << _adapterProxy->getObjProxy()->name()
// TLOGTARS("[tcp readv, " << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << ", " << _ep.desc() <<",ret " << iRet
// << ", fail! errno:" << errno << "," << strerror(errno) << ",close]" << endl);
@ -863,7 +863,7 @@ int TcpTransceiver::send(const void* buf, uint32_t len, uint32_t flag)
// return 0;
// }
// TLOGTARS("[TARS][tcp readv," << _adapterProxy->getObjProxy()->name()
// TLOGTARS("[tcp readv," << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << "," << _ep.desc() << ",ret:" << iRet << "]" << endl);
// return iRet;
@ -881,7 +881,7 @@ int TcpTransceiver::send(const void* buf, uint32_t len, uint32_t flag)
// if (iRet == 0 || (iRet < 0 && errno != EAGAIN))
// {
// TLOGTARS("[TARS][tcp recv objname:" << _adapterProxy->getObjProxy()->name()
// TLOGTARS("[tcp recv objname:" << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << ", " << _ep.desc() <<",ret " << iRet
// << ", fail! errno:" << errno << "," << strerror(errno) << ",close]" << endl);
@ -890,7 +890,7 @@ int TcpTransceiver::send(const void* buf, uint32_t len, uint32_t flag)
// return 0;
// }
// TLOGTARS("[TARS][tcp recv objname:" << _adapterProxy->getObjProxy()->name()
// TLOGTARS("[tcp recv objname:" << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << ",desc:" << _ep.desc() << ",ret:" << iRet << "]" << endl);
// return iRet;
@ -907,7 +907,7 @@ int TcpTransceiver::recv(void* buf, uint32_t len, uint32_t flag)
if (iRet == 0 || (iRet < 0 && !TC_Socket::isPending()))
{
TLOGTARS("[TARS][tcp recv, " << _adapterProxy->getObjProxy()->name()
TLOGTARS("[tcp recv, " << _adapterProxy->getObjProxy()->name()
<< ",fd:" << _fd << ", " << _ep.desc() <<", ret:" << iRet
<< ", fail! errno:" << TC_Exception::getSystemCode() << "," << TC_Exception::parseError(TC_Exception::getSystemCode()) << ",close]" << endl);
@ -922,7 +922,7 @@ int TcpTransceiver::recv(void* buf, uint32_t len, uint32_t flag)
_adapterProxy->getObjProxy()->getCommunicatorEpoll()->modFd(_fd, &_fdInfo, EPOLLIN | EPOLLOUT);
}
#endif
TLOGTARS("[TARS][tcp recv," << _adapterProxy->getObjProxy()->name()
TLOGTARS("[tcp recv," << _adapterProxy->getObjProxy()->name()
<< ",fd:" << _fd << "," << _ep.desc() << ", len:" << iRet << "]" << endl);
return iRet;
@ -968,7 +968,7 @@ int UdpTransceiver::doResponse()
{
if ((recv = this->recv(_pRecvBuffer, DEFAULT_RECV_BUFFERSIZE, 0)) > 0)
{
TLOGTARS("[TARS][udp doResponse, " << _adapterProxy->getObjProxy()->name()
TLOGTARS("[udp doResponse, " << _adapterProxy->getObjProxy()->name()
<< ",fd:" << _fd << ",recvbuf:" << recv << "]" << endl);
_recvBuffer.clearBuffers();
@ -984,7 +984,7 @@ int UdpTransceiver::doResponse()
if(ret == TC_NetWorkBuffer::PACKET_ERR || ret == TC_NetWorkBuffer::PACKET_LESS)
{
TLOGERROR("[TARS][udp doResponse," << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << "," << _ep.desc() << ",tcp recv decode error, ret:" << ret << endl);
TLOGERROR("[udp doResponse," << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << "," << _ep.desc() << ",tcp recv decode error, ret:" << ret << endl);
break;
}
else
@ -994,13 +994,13 @@ int UdpTransceiver::doResponse()
}
catch (exception &ex)
{
TLOGERROR("[TARS][udp doResponse, " << _adapterProxy->getObjProxy()->name()
TLOGERROR("[udp doResponse, " << _adapterProxy->getObjProxy()->name()
<< ",fd:" << _fd << "," << _ep.desc()
<< ", udp recv decode error:" << ex.what() << endl);
}
catch (...)
{
TLOGERROR("[TARS][udp doResponse, " << _adapterProxy->getObjProxy()->name()
TLOGERROR("[udp doResponse, " << _adapterProxy->getObjProxy()->name()
<< ",fd:" << _fd << "," << _ep.desc()
<< ", udp recv decode error." << endl);
}
@ -1022,7 +1022,7 @@ int UdpTransceiver::send(const void* buf, uint32_t len, uint32_t flag)
{
if(!TC_Socket::isPending())
{
TLOGERROR("[TARS][udp send " << _adapterProxy->getObjProxy()->name()
TLOGERROR("[udp send " << _adapterProxy->getObjProxy()->name()
<< ",fd:" << _fd << "," << _ep.desc() << ", fail! errno:"
<< TC_Exception::getSystemCode() << ","
<< TC_Exception::parseError(TC_Exception::getSystemCode()) << ",close]" << endl);
@ -1035,7 +1035,7 @@ int UdpTransceiver::send(const void* buf, uint32_t len, uint32_t flag)
}
else if(iRet>0 && iRet != (int)len)
{
TLOGERROR("[TARS][udp send, " << _adapterProxy->getObjProxy()->name()
TLOGERROR("[udp send, " << _adapterProxy->getObjProxy()->name()
<< ",fd:" << _fd << "," << _ep.desc() << ", send error."
<< ", len:" << len << ", sendLen:" << iRet << endl);
//udp只发一次 发送一半也算全部发送成功
@ -1052,7 +1052,7 @@ int UdpTransceiver::recv(void* buf, uint32_t len, uint32_t flag)
if (iRet < 0 && !TC_Socket::isPending())
{
TLOGERROR("[TARS][udp recv " << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << ","
TLOGERROR("[udp recv " << _adapterProxy->getObjProxy()->name() << ",fd:" << _fd << ","
<< _ep.desc() << ", fail! errno:" << TC_Exception::getSystemCode() << ","
<< TC_Exception::parseError(TC_Exception::getSystemCode()) << ",close]" << endl);
@ -1077,7 +1077,7 @@ int UdpTransceiver::recv(void* buf, uint32_t len, uint32_t flag)
// {
// if ((recv = this->recv(_recvBuffer, DEFAULT_RECV_BUFFERSIZE, 0)) > 0)
// {
// TLOGTARS("[TARS][udp doResponse objname:" << _adapterProxy->getObjProxy()->name()
// TLOGTARS("[udp doResponse objname:" << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << ",recvbuf:" << recv << "]" << endl);
// try
@ -1086,13 +1086,13 @@ int UdpTransceiver::recv(void* buf, uint32_t len, uint32_t flag)
// }
// catch (exception &ex)
// {
// TLOGERROR("[TARS][udp doResponse, " << _adapterProxy->getObjProxy()->name()
// TLOGERROR("[udp doResponse, " << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << ",desc:" << _ep.desc()
// << ", udp recv decode error:" << ex.what() << endl);
// }
// catch (...)
// {
// TLOGERROR("[TARS][udp doResponse, " << _adapterProxy->getObjProxy()->name()
// TLOGERROR("[udp doResponse, " << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << ",desc:" << _ep.desc()
// << ", udp recv decode error." << endl);
// }
@ -1116,7 +1116,7 @@ int UdpTransceiver::recv(void* buf, uint32_t len, uint32_t flag)
// {
// if(errno != EAGAIN)
// {
// TLOGERROR("[TARS][udp send objname:" << _adapterProxy->getObjProxy()->name()
// TLOGERROR("[udp send objname:" << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << ",desc:" << _ep.desc()
// << ", fail! errno:" << errno << "," << strerror(errno) << ",close]" << endl);
@ -1129,7 +1129,7 @@ int UdpTransceiver::recv(void* buf, uint32_t len, uint32_t flag)
// }
// else if(iRet > 0 && iRet != (int)len)
// {
// TLOGERROR("[TARS][udp send objname:" << _adapterProxy->getObjProxy()->name()
// TLOGERROR("[udp send objname:" << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << "," << _ep.desc() << ", send error."
// << ", len:" << len << ", sendLen:" << iRet << endl);
@ -1151,7 +1151,7 @@ int UdpTransceiver::recv(void* buf, uint32_t len, uint32_t flag)
// if (iRet < 0 && errno != EAGAIN)
// {
// TLOGERROR("[TARS][udp recv objname:" << _adapterProxy->getObjProxy()->name()
// TLOGERROR("[udp recv objname:" << _adapterProxy->getObjProxy()->name()
// << ",fd:" << _fd << ",desc:" << _ep.desc()
// << ", fail! errno:" << errno << "," << strerror(errno) << ",close]" << endl);

2
servant/protocol

@ -1 +1 @@
Subproject commit 36e90d1dc9adc1f8b27377f4f6500400b49d3653
Subproject commit 790bef8cbc1ba2c1b9adabf9a68ab322b9a3a119

2
servant/servant/ServantProxy.h
View File

@ -798,7 +798,7 @@ private:
* @param active
* @param inactive
*/
void onNotifyEndpoints(const set<EndpointInfo> & active,const set<EndpointInfo> & inactive);
void onNotifyEndpoints(size_t netThreadSeq, const set<EndpointInfo> & active,const set<EndpointInfo> & inactive);
/**
* 端口不活跃

17
tools/tars2cpp/tars2cpp.cpp
View File

@ -225,13 +225,22 @@ string Tars2Cpp::writeToJson(const TypeIdPtr& pPtr) const
}
else if (pPtr->getTypePtr()->isArray())
{
/*
s << TAB << "p->value[\"" << pPtr->getId() << "\"] = " + _namespace + "::JsonOutput::writeJson((const "
<< tostr(pPtr->getTypePtr()) << " *)" << pPtr->getId() << "Len" << ");" << endl;
*/
s << TAB << "p->value[\"" << pPtr->getId() << "\"] = " + _namespace + "::JsonOutput::writeJson((const "
<< tostr(pPtr->getTypePtr()) << " *)" << pPtr->getId() << ", "<< pPtr->getId() << "Len" << ");" << endl;
}
else if (pPtr->getTypePtr()->isPointer())
{
/*
s << TAB << "p->value[\"" << pPtr->getId() << "\"] = " + _namespace + "::JsonOutput::writeJson((const "
<< tostr(pPtr->getTypePtr()) << " )" << pPtr->getId() << "Len" << ");" << endl;
*/
s << TAB << "p->value[\"" << pPtr->getId() << "\"] = " + _namespace + "::JsonOutput::writeJson((const "
<< tostr(pPtr->getTypePtr()) << " )" << pPtr->getId() << ", "<< pPtr->getId() << "Len" << ");" << endl;
}
else
{
@ -290,7 +299,8 @@ string Tars2Cpp::readFromJson(const TypeIdPtr& pPtr, bool bIsRequire) const
if (pPtr->getTypePtr()->isArray())
{
s << TAB << _namespace + "::JsonInput::readJson(" << pPtr->getId() << "Len" << ",pObj->value[\"" << pPtr->getId() << "\"]" << getSuffix(pPtr);
//s << TAB << _namespace + "::JsonInput::readJson(" << pPtr->getId() << "Len" << ",pObj->value[\"" << pPtr->getId() << "\"]" << getSuffix(pPtr);
s << TAB << _namespace + "::JsonInput::readJson(" << pPtr->getId() << ", " << "sizeof("+pPtr->getId()+")" << ", " << pPtr->getId() << "Len" << ", pObj->value[\"" << pPtr->getId() << "\"]";
}
else if (pPtr->getTypePtr()->isPointer())
{
@ -381,7 +391,8 @@ string Tars2Cpp::readFrom(const TypeIdPtr& pPtr, bool bIsRequire) const
}
else if (pPtr->getTypePtr()->isArray())
{
s << TAB << "_is.read(" << pPtr->getId() << ", " << getSuffix(pPtr) << ", " << pPtr->getId() << "Len";
//s << TAB << "_is.read(" << pPtr->getId() << ", " << getSuffix(pPtr) << ", " << pPtr->getId() << "Len";
s << TAB << "_is.read(" << pPtr->getId() << ", " << "sizeof("+pPtr->getId()+")" << ", " << pPtr->getId() << "Len";
}
else if (pPtr->getTypePtr()->isPointer())
{
@ -3260,4 +3271,4 @@ StructPtr Tars2Cpp::findStruct(const ContextPtr& pPtr, const string& id)
// s << endl;
// s << "#endif" << endl;
//
//

2218
tools/tarsparse/tars.lex.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

3080
tools/tarsparse/tars.tab.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

94
tools/tarsparse/tars.tab.hpp Normal file
View File

@ -0,0 +1,94 @@
/* A Bison parser, made by GNU Bison 3.0.4. */
/* Bison interface for Yacc-like parsers in C
Copyright (C) 1984, 1989-1990, 2000-2015 Free Software Foundation, Inc.
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 3 of the License, 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. If not, see <http://www.gnu.org/licenses/>. */
/* As a special exception, you may create a larger work that contains
part or all of the Bison parser skeleton and distribute that work
under terms of your choice, so long as that work isn't itself a
parser generator using the skeleton or a modified version thereof
as a parser skeleton. Alternatively, if you modify or redistribute
the parser skeleton itself, you may (at your option) remove this
special exception, which will cause the skeleton and the resulting
Bison output files to be licensed under the GNU General Public
License without this special exception.
This special exception was added by the Free Software Foundation in
version 2.2 of Bison. */
#ifndef YY_YY_TARS_TAB_HPP_INCLUDED
# define YY_YY_TARS_TAB_HPP_INCLUDED
/* Debug traces. */
#ifndef YYDEBUG
# define YYDEBUG 1
#endif
#if YYDEBUG
extern int yydebug;
#endif
/* Token type. */
#ifndef YYTOKENTYPE
# define YYTOKENTYPE
enum yytokentype
{
TARS_VOID = 258,
TARS_STRUCT = 259,
TARS_BOOL = 260,
TARS_BYTE = 261,
TARS_SHORT = 262,
TARS_INT = 263,
TARS_DOUBLE = 264,
TARS_FLOAT = 265,
TARS_LONG = 266,
TARS_STRING = 267,
TARS_VECTOR = 268,
TARS_MAP = 269,
TARS_NAMESPACE = 270,
TARS_INTERFACE = 271,
TARS_IDENTIFIER = 272,
TARS_OUT = 273,
TARS_OP = 274,
TARS_KEY = 275,
TARS_ROUTE_KEY = 276,
TARS_REQUIRE = 277,
TARS_OPTIONAL = 278,
TARS_CONST_INTEGER = 279,
TARS_CONST_FLOAT = 280,
TARS_FALSE = 281,
TARS_TRUE = 282,
TARS_STRING_LITERAL = 283,
TARS_SCOPE_DELIMITER = 284,
TARS_CONST = 285,
TARS_ENUM = 286,
TARS_UNSIGNED = 287,
BAD_CHAR = 288
};
#endif
/* Value type. */
#if ! defined YYSTYPE && ! defined YYSTYPE_IS_DECLARED
typedef int YYSTYPE;
# define YYSTYPE_IS_TRIVIAL 1
# define YYSTYPE_IS_DECLARED 1
#endif
extern YYSTYPE yylval;
int yyparse (void);
#endif /* !YY_YY_TARS_TAB_HPP_INCLUDED */

2
util/src/tc_epoll_server.cpp
View File

@ -686,7 +686,7 @@ TC_EpollServer::Connection::Connection(TC_EpollServer::BindAdapter *pBindAdapter
, _bEmptyConn(true)
, _pRecvBuffer(NULL)
, _nRecvBufferSize(DEFAULT_RECV_BUFFERSIZE)
, _authInit(false)
{
assert(fd != -1);