123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602 |
- /**
- * Tencent is pleased to support the open source community by making Tars available.
- *
- * Copyright (C) 2016THL A29 Limited, a Tencent company. All rights reserved.
- *
- * Licensed under the BSD 3-Clause License (the "License"); you may not use this file except
- * in compliance with the License. You may obtain a copy of the License at
- *
- * https://opensource.org/licenses/BSD-3-Clause
- *
- * Unless required by applicable law or agreed to in writing, software distributed
- * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
- * CONDITIONS OF ANY KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations under the License.
- */
- #include "servant/ServantProxy.h"
- #include "servant/ServantHandle.h"
- #include "servant/StatReport.h"
- #include "servant/Application.h"
- #include "servant/BaseF.h"
- #include "servant/CommunicatorEpoll.h"
- #include "servant/EndpointManager.h"
- #include "servant/Message.h"
- #include "servant/ObjectProxy.h"
- #include "servant/RemoteLogger.h"
- namespace tars
- {
- shared_ptr<ServantProxyThreadData::Immortal> ServantProxyThreadData::g_immortal;
- thread_local shared_ptr<ServantProxyThreadData> ServantProxyThreadData::g_sp;
- unsigned int ServantProxyThreadData::_traceParamMaxLen = 1; // 默认1K
- ///////////////////////////////////////////////////////////////
- SeqManager::SeqManager(uint16_t iNum)
- {
- assert(iNum < MAX_UNSIGN_SHORT);
- assert(iNum > 0);
- _p = NULL;
- _p = new SeqInfo[iNum];
- assert(_p);
- //0xff表示结束
- _free = 0;
- _freeTail = iNum - 1;
- for (uint16_t i = 0; i < (uint16_t)iNum; i++)
- {
- _p[i].free = true;
- _p[i].next = i + 1;
- }
- _p[iNum - 1].next = MAX_UNSIGN_SHORT;
- _num = iNum;
- }
- SeqManager ::~SeqManager()
- {
- if (_p)
- {
- delete[] _p;
- _p = NULL;
- }
- }
- uint16_t SeqManager::get()
- {
- TC_LockT<TC_SpinLock> lock(_mutex);
- assert(_free != MAX_UNSIGN_SHORT);
- assert(_p[_free].free);
- uint16_t buf = _free;
- if (_free == _freeTail)
- {
- assert(_p[buf].next == MAX_UNSIGN_SHORT);
- _freeTail = MAX_UNSIGN_SHORT;
- }
- _free = _p[buf].next;
- _p[buf].free = false;
- return buf;
- }
- void SeqManager::del(uint16_t iSeq)
- {
- TC_LockT<TC_SpinLock> lock(_mutex);
- assert(iSeq < _num);
- assert(!_p[iSeq].free);
- _p[iSeq].next = MAX_UNSIGN_SHORT;
- if (MAX_UNSIGN_SHORT == _freeTail)
- {
- _free = iSeq;
- }
- else
- {
- _p[_freeTail].next = iSeq;
- }
- _p[iSeq].free = true;
- _freeTail = iSeq;
- }
- ///////////////////////////////////////////////////////////////
- ServantProxyThreadData::Immortal::Immortal()
- {
- _pSeq.reset(new SeqManager(MAX_CLIENT_NOTIFYEVENT_NUM));
- }
- ServantProxyThreadData::Immortal::~Immortal()
- {
- _pSeq.reset();
- }
- void ServantProxyThreadData::Immortal::add(ServantProxyThreadData* data)
- {
- TC_LockT<TC_ThreadMutex> lock(_mutex);
- _sp_list.insert(data);
- }
- void ServantProxyThreadData::Immortal::erase(ServantProxyThreadData* data)
- {
- TC_LockT<TC_ThreadMutex> lock(_mutex);
- _sp_list.erase(data);
- }
- unordered_set<ServantProxyThreadData *> ServantProxyThreadData::Immortal::getList()
- {
- TC_LockT<TC_ThreadMutex> lock(_mutex);
- return _sp_list;
- }
- void ServantProxyThreadData::Immortal::erase(Communicator *comm)
- {
- TC_LockT<TC_ThreadMutex> lock(_mutex);
- for(auto it : _sp_list)
- {
- (*it).erase(comm);
- }
- }
- ServantProxyThreadData::ServantProxyThreadData()
- : _reqQNo(0)
- {
- // LOG_CONSOLE_DEBUG << endl;
- _sp_immortal = g_immortal;
- _reqQNo = _sp_immortal->getSeqManager()->get();
- _sp_immortal->add(this);
- }
- ServantProxyThreadData::~ServantProxyThreadData()
- {
- // LOG_CONSOLE_DEBUG << endl;
- try
- {
- // TC_LockT<TC_SpinLock> lock(_mutex);
- //先释放公有的网络通信器的信息
- for(auto it = _communicatorEpollInfo.begin(); it != _communicatorEpollInfo.end(); ++it)
- {
- for (auto &e: it->second->_info)
- {
- shared_ptr<CommunicatorEpoll> ce = e._communicatorEpoll.lock();
- if(ce)
- {
- ce->notifyServantThreadQuit(_reqQNo);
- }
- }
- }
- for(auto it = _schedCommunicatorEpollInfo.begin(); it != _schedCommunicatorEpollInfo.end(); ++it)
- {
- it->second->_communicator->eraseSchedCommunicatorEpoll(_reqQNo);
- }
- }
- catch (...)
- {
- }
- _sp_immortal->erase(this);
- _sp_immortal->getSeqManager()->del(_reqQNo);
- _sched.reset();
- _sp_immortal.reset();
- }
- ServantProxyThreadData* ServantProxyThreadData::getData()
- {
- static std::once_flag flag;
- std::call_once(flag, []()
- { g_immortal = std::make_shared<Immortal>(); });
- if (!g_sp)
- {
- g_sp = std::make_shared<ServantProxyThreadData>();
- }
- return g_sp.get();
- }
- void ServantProxyThreadData::deconstructor(Communicator *communicator)
- {
- if(g_immortal.use_count() > 0)
- {
- g_immortal->erase(communicator);
- }
- }
- void ServantProxyThreadData::erase(Communicator *communicator)
- {
- TC_LockT<TC_ThreadMutex> lock(_mutex);
- _communicatorEpollInfo.erase(communicator);
- _schedCommunicatorEpollInfo.erase(communicator);
- }
- shared_ptr<ServantProxyThreadData::CommunicatorEpollInfo> ServantProxyThreadData::getCommunicatorEpollInfo(Communicator *communicator)
- {
- TC_LockT<TC_ThreadMutex> lock(_mutex);
- auto it = _communicatorEpollInfo.find(communicator);
- if(it != _communicatorEpollInfo.end())
- {
- return it->second;
- }
- auto info = std::make_shared<CommunicatorEpollInfo>();
- _communicatorEpollInfo.insert(std::make_pair(communicator, info));
- return info;
- }
- shared_ptr<ServantProxyThreadData::SchedCommunicatorEpollInfo> ServantProxyThreadData::getSchedCommunicatorEpollInfo(Communicator *communicator)
- {
- TC_LockT<TC_ThreadMutex> lock(_mutex);
- auto it = _schedCommunicatorEpollInfo.find(communicator);
- if(it != _schedCommunicatorEpollInfo.end())
- {
- return it->second;
- }
- auto info = std::make_shared<SchedCommunicatorEpollInfo>();
- _schedCommunicatorEpollInfo.insert(std::make_pair(communicator, info));
- return info;
- }
- ThreadPrivateData ServantProxyThreadData::move()
- {
- ThreadPrivateData data = _data;
- //hash每次调用完成都要清掉,不用透传
- _data._hash = false;
- _data._conHash = false;
- _data._timeout = 0;
- return data;
- }
- shared_ptr<ServantProxyThreadData::CommunicatorEpollInfo> ServantProxyThreadData::addCommunicatorEpoll(const shared_ptr<CommunicatorEpoll> &ce)
- {
- auto q = std::make_shared<ReqInfoQueue>(ce->getNoSendQueueLimit());
- ServantProxyThreadData::CommunicatorEpollReqQueueInfo epollReqQueueInfo;
- epollReqQueueInfo._reqQueue = q;
- epollReqQueueInfo._communicatorEpoll = ce;
- auto info = getCommunicatorEpollInfo(ce->getCommunicator());
- info->_info.push_back(epollReqQueueInfo);
- ce->initNotify(this->_reqQNo, q);
- return info;
- }
- ///////////////////////////////////////////////////////////////////////////////////////////
- ServantProxyCallback::ServantProxyCallback()
- : _bNetThreadProcess(false)
- {
- }
- int ServantProxyCallback::dispatch(ReqMessagePtr msg)
- {
- return onDispatch(msg);
- }
- ///////////////////////////////////////////////////////////////////////////////////////////
- int HttpServantProxyCallback::onDispatch(ReqMessagePtr msg)
- {
- if (msg->response->iRet != TARSSERVERSUCCESS)
- {
- return onDispatchException(msg->request, *msg->response);
- }
- return onDispatchResponse(msg->request, *msg->response);
- }
- HttpServantProxyCallback::HttpServantProxyCallback(const HttpCallbackPtr &cb)
- : _httpCb(cb)
- {
- }
- int HttpServantProxyCallback::onDispatchException(const RequestPacket &request, const ResponsePacket &response)
- {
- if (_httpCb)
- return _httpCb->onHttpResponseException(response.iRet);
- return 0;
- }
- int HttpServantProxyCallback::onDispatchResponse(const RequestPacket &request, const ResponsePacket &response)
- {
- assert(response.sBuffer.size() == sizeof(shared_ptr<TC_HttpResponse>));
- shared_ptr<TC_HttpResponse> rsp = *(shared_ptr<TC_HttpResponse> *)(response.sBuffer.data());
- if (_httpCb)
- return _httpCb->onHttpResponse(rsp);
- return 0;
- }
- ///////////////////////////////////////////////////////////////
- void coroWhenAll(const CoroParallelBasePtr &ptr)
- {
- if(!ptr->checkAllReqSend())
- {
- 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");
- }
- //线程私有数据
- ServantProxyThreadData *pSptd = ServantProxyThreadData::getData();
- assert(pSptd != NULL);
- if (!pSptd->_sched)
- {
- TLOGERROR("[coroWhenAll no open coroutine mode]"<<endl);
- throw TarsUseCoroException("coroWhenAll not open coroutine mode");
- }
- pSptd->_sched->yield(false);
- vector<ReqMessage *> vMsg = ptr->getAllReqMessage();
- for (size_t i = 0; i < vMsg.size(); ++i)
- {
- ReqMessagePtr msgPtr = vMsg[i];
- vMsg[i]->callback->dispatch(msgPtr);
- }
- }
- ///////////////////////////////////////////////////////////////
- string ServantProxy::STATUS_DYED_KEY = "STATUS_DYED_KEY";
- string ServantProxy::STATUS_GRID_KEY = "STATUS_GRID_KEY";
- string ServantProxy::STATUS_RESULT_CODE = "STATUS_RESULT_CODE";
- string ServantProxy::STATUS_RESULT_DESC = "STATUS_RESULT_DESC";
- string ServantProxy::STATUS_SETNAME_VALUE = "STATUS_SETNAME_VALUE";
- string ServantProxy::STATUS_TRACE_KEY = "STATUS_TRACE_KEY";
- ////////////////////////////////////
- ServantProxy::ServantProxy(Communicator *pCommunicator, const string &name, const string &setName)
- : _communicator(pCommunicator), _syncTimeout(DEFAULT_SYNCTIMEOUT), _asyncTimeout(DEFAULT_ASYNCTIMEOUT), _id(0), _masterFlag(false), _minTimeout(100)
- {
- _proxyProtocol.requestFunc = ProxyProtocol::tarsRequest;
- _proxyProtocol.responseFunc = ProxyProtocol::tarsResponse;
- //在每个公有网络线程对象中创建ObjectProxy
- for (size_t i = 0; i < _communicator->getCommunicatorEpollNum(); ++i)
- {
- _communicator->getCommunicatorEpoll(i)->createObjectProxy(this, name, setName);
- }
- //用第一个ObjectProxy返回数据
- _objectProxy = this->getObjectProxy(0);
- _endpointInfo.reset(new EndpointManagerThread(_communicator, _objectProxy->name()));
- _minTimeout = pCommunicator->getMinTimeout();
- if (_minTimeout < 1)
- {
- _minTimeout = 1;
- }
- }
- void ServantProxy::tars_initialize(bool rootServant)
- {
- //等ServantProxy完全创建完以后, 再创建Object
- for (size_t i = 0; i < _communicator->getCommunicatorEpollNum(); ++i)
- {
- this->getObjectProxy(i)->initialize(rootServant);
- }
- }
- ServantProxy::~ServantProxy()
- {
- }
- const string &ServantProxy::tars_name() const
- {
- return _objectProxy->name();
- }
- string ServantProxy::tars_full_name() const
- {
- string name = _objectProxy->name();
- if(!_objectProxy->hash().empty())
- {
- name = name + "#" + _objectProxy->hash();
- }
- if(!_objectProxy->address().empty())
- {
- name = name + "@" + _objectProxy->address();
- }
- return name;
- }
- const string& ServantProxy::tars_setName() const
- {
- return _objectProxy->getInvokeSetName();
- }
- ObjectProxy *ServantProxy::getObjectProxy(size_t netThreadSeq)
- {
- return _communicator->getCommunicatorEpoll(netThreadSeq)->servantToObjectProxy(this);
- }
- void ServantProxy::forEachObject(std::function<void(ObjectProxy *)> func)
- {
- for (size_t i = 0; i < _communicator->getCommunicatorEpollNum(); ++i)
- {
- ObjectProxy *objectProxy = _communicator->getCommunicatorEpoll(i)->servantToObjectProxy(this);
- assert(objectProxy != NULL);
- try
- {
- func(objectProxy);
- }
- catch (...)
- {
- }
- }
- }
- void ServantProxy::tars_reconnect(int second)
- {
- forEachObject([=](ObjectProxy *o) { o->reconnect(second); });
- }
- TC_Endpoint ServantProxy::tars_invoke_endpoint()
- {
- ServantProxyThreadData *td = ServantProxyThreadData::getData();
- if (td)
- {
- return TC_Endpoint(td->_data._szHost);
- }
- return TC_Endpoint();
- }
- void ServantProxy::tars_set_proxy(ServantProxy::SERVANT_PROXY type, const TC_Endpoint &ep, const string &user, const string &pass)
- {
- _proxyBaseInfo = std::make_shared<TC_ProxyInfo::ProxyBaseInfo>();
- _proxyBaseInfo->ep = ep;
- _proxyBaseInfo->user = user;
- _proxyBaseInfo->pass = pass;
- switch (type)
- {
- case PROXY_SOCK4:
- _proxyBaseInfo->type = TC_ProxyInfo::eProxy_Type_Sock4;
- break;
- case PROXY_SOCK5:
- _proxyBaseInfo->type = TC_ProxyInfo::eProxy_Type_Sock5;
- break;
- case PROXY_HTTP:
- _proxyBaseInfo->type = TC_ProxyInfo::eProxy_Type_Http;
- break;
- default:
- assert(false);
- }
- }
- void ServantProxy::tars_timeout(int msecond)
- {
- //保护,超时时间不能小于_minTimeout毫秒
- _syncTimeout = (msecond < _minTimeout) ? _minTimeout : msecond;
- }
- int ServantProxy::tars_timeout() const
- {
- return _syncTimeout;
- }
- int ServantProxy::tars_connect_timeout() const
- {
- return _connTimeout;
- }
- void ServantProxy::tars_connect_timeout(int conTimeout)
- {
- if(conTimeout < 100)
- {
- conTimeout = 100;
- }
- _connTimeout = conTimeout;
- }
- void ServantProxy::tars_async_timeout(int msecond)
- {
- TC_LockT<TC_ThreadMutex> lock(*this);
- //保护,超时时间不能小于_minTimeout毫秒
- _asyncTimeout = (msecond < _minTimeout) ? _minTimeout : msecond;
- }
- int ServantProxy::tars_async_timeout() const
- {
- return _asyncTimeout;
- }
- void ServantProxy::tars_set_custom_callback(ServantProxy::custom_callback callback)
- {
- _callback = callback;
- }
- void ServantProxy::tars_enable_callback_hash()
- {
- _callbackHash = true;
- }
- void ServantProxy::tars_connection_serial(int connectionSerial)
- {
- assert(!_rootPrx);
- _connectionSerial = connectionSerial;
- }
- int ServantProxy::tars_connection_serial() const
- {
- if(_rootPrx) {
- return _rootPrx->tars_connection_serial();
- }
- return _connectionSerial;
- }
- void ServantProxy::tars_set_protocol(SERVANT_PROTOCOL protocol, int connectionSerial)
- {
- ProxyProtocol proto;
- switch(protocol)
- {
- case PROTOCOL_HTTP1:
- proto.requestFunc = ProxyProtocol::http1Request;
- proto.responseFunc = ProxyProtocol::http1Response;
- if(connectionSerial <= 0)
- connectionSerial = DEFAULT_CONNECTION_SERIAL;
- break;
- #if TARS_HTTP2
- case PROTOCOL_HTTP2:
- proto.requestFunc = ProxyProtocol::http2Request;
- proto.responseFunc = ProxyProtocol::http2Response;
- connectionSerial = 0;
- break;
- case PROTOCOL_GRPC:
- proto.requestFunc = ProxyProtocol::grpcRequest;
- proto.responseFunc = ProxyProtocol::grpcResponse;
- connectionSerial = 0;
- break;
- #endif
- case PROTOCOL_TARS:
- default:
- proto.requestFunc = ProxyProtocol::tarsRequest;
- proto.responseFunc = ProxyProtocol::tarsResponse;
- break;
- }
- tars_set_protocol(proto, connectionSerial);
- }
- void ServantProxy::tars_set_protocol(const ProxyProtocol& protocol, int connectionSerial)
- {
- TC_LockT<TC_ThreadMutex> lock(*this);
- _proxyProtocol = protocol;
- _connectionSerial = connectionSerial;
- }
- const ProxyProtocol &ServantProxy::tars_get_protocol() const
- {
- return _proxyProtocol;
- }
- vector<ServantProxy::SocketOpt> ServantProxy::tars_get_sockopt() const
- {
- TC_LockT<TC_ThreadMutex> lock(*this);
- return _socketOpts;
- }
- void ServantProxy::tars_set_sockopt(int level, int optname, const void *optval, SOCKET_LEN_TYPE optlen)
- {
- TC_LockT<TC_ThreadMutex> lock(*this);
- SocketOpt socketOpt;
- socketOpt.level = level;
- socketOpt.optname = optname;
- socketOpt.optval = optval;
- socketOpt.optlen = optlen;
- _socketOpts.push_back(socketOpt);
- }
- void ServantProxy::tars_set_check_timeout(const CheckTimeoutInfo& checkTimeoutInfo)
- {
- TC_LockT<TC_ThreadMutex> lock(*this);
- _checkTimeoutInfo = checkTimeoutInfo;
- }
- CheckTimeoutInfo ServantProxy::tars_get_check_timeout()
- {
- TC_LockT<TC_ThreadMutex> lock(*this);
- return _checkTimeoutInfo;
- }
- void ServantProxy::tars_ping()
- {
- map<string, string> m;
- map<string, string> s;
- TarsOutputStream<BufferWriterVector> os;
- tars_invoke(TARSNORMAL, "tars_ping", os, m, s);
- }
- void ServantProxy::tars_async_ping()
- {
- map<string, string> m;
- map<string, string> s;
- TarsOutputStream<BufferWriterVector> os;
- tars_invoke_async(TARSONEWAY, "tars_ping", os, m, s, NULL);
- }
- ServantProxy* ServantProxy::tars_hash(int64_t key)
- {
- ServantProxyThreadData *pSptd = ServantProxyThreadData::getData();
- assert(pSptd != NULL);
- pSptd->_data._hash = true;
- pSptd->_data._hashCode = key;
- return this;
- }
- //ServantProxy* ServantProxy::tars_direct()
- //{
- // ServantProxyThreadData *pSptd = ServantProxyThreadData::getData();
- //
- // pSptd->_direct = true;
- // return this;
- //}
- ServantProxy* ServantProxy::tars_consistent_hash(int64_t key)
- {
- ServantProxyThreadData *pSptd = ServantProxyThreadData::getData();
- assert(pSptd != NULL);
- pSptd->_data._hash = true;
- pSptd->_data._conHash = true;
- pSptd->_data._hashCode = key;
- return this;
- }
- ServantProxy* ServantProxy::tars_open_trace(bool traceParams)
- {
- ServantProxyThreadData *pSptd = ServantProxyThreadData::getData();
- assert(pSptd != NULL);
- if (traceParams)
- {
- pSptd->openTrace(15);
- }
- else
- {
- pSptd->openTrace();
- }
- return this;
- }
- void ServantProxy::tars_clear_hash()
- {
- }
- ServantProxy* ServantProxy::tars_set_timeout(int msecond)
- {
- ServantProxyThreadData *pSptd = ServantProxyThreadData::getData();
- assert(pSptd != NULL);
- // pSptd->_hasTimeout = true;
- pSptd->_data._timeout = msecond;
- return this;
- }
- uint32_t ServantProxy::tars_gen_requestid()
- {
- uint32_t i = ++_id;
- if (i == 0)
- {
- i = ++_id;
- }
- return i;
- }
- void ServantProxy::tars_set_push_callback(const ServantProxyCallbackPtr & cb)
- {
- _pushCallback = cb;
- }
- ServantProxyCallbackPtr ServantProxy::tars_get_push_callback()
- {
- return _pushCallback;
- }
- void ServantProxy::invoke(ReqMessage *msg, bool bCoroAsync)
- {
- //线程私有数据
- ServantProxyThreadData *pSptd = ServantProxyThreadData::getData();
- assert(pSptd != NULL);
- //协程调用方式, 启用协程
- if(bCoroAsync && TC_CoroutineScheduler::scheduler() && !pSptd->_sched)
- {
- pSptd->_sched = TC_CoroutineScheduler::scheduler();
- }
- msg->data = pSptd->move();
- // 调用链追踪透传
- msg->bTraceCall = pSptd->_traceCall;
- msg->sTraceKey = pSptd->getTraceKey(true);
- if (msg->data._dyeing)
- {
- TLOGTARS("[ServantProxy::invoke, set dyeing, key=" << pSptd->_data._dyeingKey << endl);
- }
- msg->data._cookie = pSptd->_data._cookie;
- // #ifdef TARS_OPENTRACKING
- // msg->trackInfoMap = pSptd->_trackInfoMap;
- // #endif
- //
- //设置超时时间
- msg->request.iTimeout = (ReqMessage::SYNC_CALL == msg->eType) ? _syncTimeout : _asyncTimeout;
- //判断是否针对接口级设置超时
- if (msg->data._timeout > 0)
- {
- msg->request.iTimeout = msg->data._timeout;
- }
- else
- {
- msg->request.iTimeout = (ReqMessage::SYNC_CALL == msg->eType) ? _syncTimeout : _asyncTimeout;
- }
- shared_ptr<ReqInfoQueue> pReqQ;
- //选择网络线程
- selectNetThreadInfo(pSptd, msg->pObjectProxy, pReqQ);
- //如果是按set规则调用
- if (msg->pObjectProxy && msg->pObjectProxy->isInvokeBySet())
- {
- SET_MSG_TYPE(msg->request.iMessageType, TARSMESSAGETYPESETNAME);
- msg->request.status[ServantProxy::STATUS_SETNAME_VALUE] = msg->pObjectProxy->getInvokeSetName();
- TLOGTARS("[ServantProxy::invoke, " << msg->request.sServantName << ", invoke with set," << msg->pObjectProxy->getInvokeSetName() << "]" << endl);
- }
- assert(msg->pMonitor == NULL);
- if (msg->eType == ReqMessage::SYNC_CALL)
- {
- if (pSptd->_sched)
- {
- msg->sched = pSptd->_sched;
- msg->iCoroId = pSptd->_sched->getCoroutineId();
- }
- else
- {
- //同步调用 new 一个ReqMonitor
- msg->pMonitor = new ReqMonitor(msg);
- }
- }
- else if (msg->eType == ReqMessage::ASYNC_CALL)
- {
- //是否是协程的并行请求
- if (bCoroAsync)
- {
- if (pSptd->_sched)
- {
- CoroParallelBasePtr coroPtr = msg->callback->getCoroParallelBasePtr();
- if (coroPtr)
- {
- coroPtr->incReqCount();
- msg->sched = pSptd->_sched;
- msg->iCoroId = pSptd->_sched->getCoroutineId();
- }
- else
- {
- TLOGERROR("[ServantProxy::invoke use coroutine's callback not set CoroParallelBasePtr]" << endl);
- delete msg;
- msg = NULL;
- throw TarsUseCoroException("use coroutine's callback not set CoroParallelBasePtr");
- }
- }
- else
- {
- TLOGERROR("[ServantProxy::invoke coroutine mode invoke not open]" << endl);
- delete msg;
- msg = NULL;
- throw TarsUseCoroException("coroutine mode invoke not open");
- }
- }
- }
- //通知网络线程
- bool bEmpty = false;
- bool bSync = (msg->eType == ReqMessage::SYNC_CALL);
- if (!pReqQ->push_back(msg, bEmpty))
- {
- TLOGERROR("[ServantProxy::invoke msgQueue push_back error thread seq:" << pSptd->_reqQNo << "]" << endl);
- delete msg;
- msg = NULL;
- throw TarsClientQueueException("client queue full");
- }
- if (msg->sched)
- {
- // LOG_CONSOLE_DEBUG << "in sched handle: " << this << ", " << msg->request.sServantName << endl;
- //协程中, 直接发包了
- msg->pObjectProxy->getCommunicatorEpoll()->handle(pSptd->_reqQNo);
- }
- else
- {
- msg->pObjectProxy->getCommunicatorEpoll()->notify(pSptd->_reqQNo);
- }
- if (bSync)
- {
- if (!msg->sched)
- {
- assert(msg->pMonitor);
- msg->pMonitor->wait();
- if(!msg->pMonitor->bMonitorFin)
- {
- TLOGERROR("[ServantProxy::invoke communicator terminate]" << endl);
- throw TarsCommunicatorException("communicator terminate");
- }
- }
- else
- {
- assert(!msg->sched->isMainCoroutine());
- msg->sched->yield(false);
- }
- //判断eStatus来判断状态
- // assert(msg->eStatus != ReqMessage::REQ_REQ);
- if(msg->adapter) {
- pSptd->_data._szHost = msg->adapter->endpoint().desc();
- }
- if(msg->eStatus == ReqMessage::REQ_RSP && msg->response->iRet == TARSSERVERSUCCESS)
- {
- //成功
- return;
- }
- ostringstream os;
- if (msg->eStatus == ReqMessage::REQ_TIME)
- {
- //超时
- os << "[ServantProxy::invoke timeout:" << msg->request.iTimeout;
- }
- else
- {
- os << "[ServantProxy::invoke errno:" << msg->response->iRet << ",info:" << msg->response->sResultDesc;
- }
- os << ",servant:" << msg->pObjectProxy->name() << ",func:" << msg->request.sFuncName;
- if (msg->adapter)
- os << ",adapter:" << msg->adapter->endpoint().desc();
- os << ",reqid:" << msg->request.iRequestId << "]";
- if (msg->eStatus == ReqMessage::REQ_TIME)
- {
- //超时
- delete msg;
- msg = NULL;
- throw TarsSyncCallTimeoutException(os.str());
- }
- //异常调用
- int ret = msg->response->iRet;
- delete msg;
- msg = NULL;
- TarsException::throwException(ret, os.str());
- }
- }
- //////////////////////////////////////////////////////////////////
- void ServantProxy::tars_invoke_async(char cPacketType,
- const string &sFuncName,
- TarsOutputStream<BufferWriterVector> &buf,
- const map<string, string>& context,
- const map<string, string>& status,
- const ServantProxyCallbackPtr& callback,
- bool bCoro)
- {
- ReqMessage *msg = new ReqMessage();
- msg->init(callback?ReqMessage::ASYNC_CALL:ReqMessage::ONE_WAY, this);
- msg->callback = callback;
- msg->request.iVersion = TARSVERSION;
- msg->request.cPacketType = (callback ? cPacketType : TARSONEWAY);
- msg->request.sFuncName = sFuncName;
- msg->request.sServantName = _objectProxy->name();
- buf.swap(msg->request.sBuffer);
- msg->request.context = context;
- msg->request.status = status;
- msg->request.iTimeout = _asyncTimeout;
-
- // // 在RequestPacket中的context设置主调信息
- // if(_masterFlag)
- // {
- // msg->request.context.insert(std::make_pair(TARS_MASTER_KEY,ClientConfig::ModuleName)); //TARS_MASTER_KEY clientConfig.ModuleName
- // }
- checkDye(msg->request);
- checkTrace(msg->request);
- checkCookie(msg->request);
- servant_invoke(msg, bCoro);
- }
- //////////////////////////////////////////////////////////////////
- void ServantProxy::tars_invoke_async(char cPacketType,
- const string &sFuncName,
- const vector<char> &buf,
- const map<string, string>& context,
- const map<string, string>& status,
- const ServantProxyCallbackPtr& callback,
- bool bCoro)
- {
- ReqMessage * msg = new ReqMessage();
- msg->init(callback?ReqMessage::ASYNC_CALL:ReqMessage::ONE_WAY, this);
- msg->callback = callback;
- msg->request.iVersion = TARSVERSION;
- msg->request.cPacketType = (callback ? cPacketType : TARSONEWAY);
- msg->request.sFuncName = sFuncName;
- msg->request.sServantName = _objectProxy->name();
- msg->request.sBuffer = buf;
- msg->request.context = context;
- msg->request.status = status;
- msg->request.iTimeout = _asyncTimeout;
-
- // // 在RequestPacket中的context设置主调信息
- // if(_masterFlag)
- // {
- // msg->request.context.insert(std::make_pair(TARS_MASTER_KEY,ClientConfig::ModuleName)); //TARS_MASTER_KEY clientConfig.ModuleName
- // }
- checkDye(msg->request);
- checkTrace(msg->request);
- checkCookie(msg->request);
- servant_invoke(msg, bCoro);
- }
- shared_ptr<ResponsePacket> ServantProxy::tars_invoke(char cPacketType,
- const string& sFuncName,
- const vector<char>& buf,
- const map<string, string>& context,
- const map<string, string>& status)
- // ResponsePacket& rsp)
- {
- ReqMessage *msg = new ReqMessage();
- msg->init(ReqMessage::SYNC_CALL, this);
- msg->request.iVersion = TARSVERSION;
- msg->request.cPacketType = cPacketType;
- msg->request.sFuncName = sFuncName;
- msg->request.sServantName = _objectProxy->name();
- msg->request.sBuffer = buf;
- msg->request.context = context;
- msg->request.status = status;
- msg->request.iTimeout = _syncTimeout;
- // // 在RequestPacket中的context设置主调信息
- // if(_masterFlag)
- // {
- // msg->request.context.insert(std::make_pair(TARS_MASTER_KEY,ClientConfig::ModuleName));
- // }
- checkDye(msg->request);
- checkTrace(msg->request);
- checkCookie(msg->request);
- servant_invoke(msg, false);
- shared_ptr<ResponsePacket> rsp = msg->response;
- // rsp = msg->response;
- delete msg;
- msg = NULL;
- return rsp;
- }
- shared_ptr<ResponsePacket> ServantProxy::tars_invoke(char cPacketType,
- const string& sFuncName,
- TarsOutputStream<BufferWriterVector>& buf,
- const map<string, string>& context,
- const map<string, string>& status)
- // ResponsePacket& rsp)
- {
- ReqMessage * msg = new ReqMessage();
- msg->init(ReqMessage::SYNC_CALL, this);
- msg->request.iVersion = TARSVERSION;
- msg->request.cPacketType = cPacketType;
- msg->request.sFuncName = sFuncName;
- msg->request.sServantName = _objectProxy->name();
- buf.swap(msg->request.sBuffer);
- msg->request.context = context;
- msg->request.status = status;
- msg->request.iTimeout = _syncTimeout;
- // // 在RequestPacket中的context设置主调信息
- // if(_masterFlag)
- // {
- // msg->request.context.insert(std::make_pair(TARS_MASTER_KEY,ClientConfig::ModuleName));
- // }
- checkDye(msg->request);
- checkTrace(msg->request);
- checkCookie(msg->request);
- servant_invoke(msg, false);
- shared_ptr<ResponsePacket> rsp = msg->response;
- // rsp = msg->response;
- delete msg;
- msg = NULL;
- return rsp;
- }
- //////////////////////////////////////////////////////////////////////////////
- //服务端是非tars协议,通过rpc_call调用
- void ServantProxy::rpc_call(uint32_t iRequestId,
- const string& sFuncName,
- const char* buff,
- uint32_t len,
- ResponsePacket& rsp)
- {
- ReqMessage *msg = new ReqMessage();
- msg->init(ReqMessage::SYNC_CALL, this);
- msg->bFromRpc = true;
- msg->request.sServantName = _objectProxy->name();
- msg->request.sFuncName = sFuncName;
- msg->request.iRequestId = iRequestId;
- msg->request.sBuffer.assign(buff, buff + len);
- servant_invoke(msg, false);
- rsp = *msg->response.get();
- delete msg;
- msg = NULL;
- }
- void ServantProxy::rpc_call_async(uint32_t iRequestId,
- const string& sFuncName,
- const char* buff,
- uint32_t len,
- const ServantProxyCallbackPtr& callback,
- bool bCoro)
- {
- ReqMessage *msg = new ReqMessage();
- msg->init(callback ? ReqMessage::ASYNC_CALL : ReqMessage::ONE_WAY, this);
- msg->bFromRpc = true;
- msg->callback = callback;
- msg->request.sServantName = _objectProxy->name();
- msg->request.sFuncName = sFuncName;
- msg->request.iRequestId = iRequestId;
- msg->request.sBuffer.assign(buff, buff + len);
- servant_invoke(msg, bCoro);
- }
- ServantPrx ServantProxy::getServantPrx(ReqMessage *msg)
- {
- if (_servantId == 0)
- {
- std::lock_guard<std::mutex> m(_servantMutex);
- if (_servantId == 0 && _servantList.empty())
- {
- for (int i = 1; i < _connectionSerial; ++i)
- {
- string obj = tars_name() + "#" + TC_Common::tostr(i);
- if (!_objectProxy->address().empty())
- {
- obj += "@" + _objectProxy->address();
- }
- ServantPrx prx = _communicator->stringToProxy<ServantPrx>(obj, this->tars_setName(), false);
- prx->tars_set_protocol(tars_get_protocol());
- prx->tars_connect_timeout(tars_connect_timeout());
- prx->tars_timeout(tars_timeout());
- prx->tars_async_timeout(tars_async_timeout());
- prx->_rootPrx = this;
- _servantList.push_back(prx);
- }
- }
- }
- if (_connectionSerial != 0)
- {
- assert(int(_servantList.size() + 1) == _connectionSerial);
- }
- int id = _servantId++ % (_servantList.size() + 1);
- if (id == 0)
- {
- return this;
- }
- return _servantList[(id - 1)];
- }
- void ServantProxy::travelObjectProxys(ServantProxy *prx, function<void(ObjectProxy*)> f)
- {
- vector<ObjectProxy*> objectProxys;
- size_t num = _communicator->getCommunicatorEpollNum();
- for (size_t i = 0; i < num; ++i)
- {
- auto ce = _communicator->getCommunicatorEpoll(i);
- ObjectProxy* objectProxy = ce->servantToObjectProxy(prx);
- if (objectProxy)
- {
- f(objectProxy);
- }
- }
- //协程通信器也需要
- _communicator->forEachSchedCommunicatorEpoll([&](const shared_ptr<CommunicatorEpoll>& ce)
- {
- ObjectProxy* objectProxy = ce->servantToObjectProxy(prx);
- if (objectProxy)
- {
- f(objectProxy);
- }
- });
- }
- vector<ObjectProxy*> ServantProxy::getObjectProxys()
- {
- vector<ObjectProxy*> objectProxys;
- //更新子servant proxy的地址
- for (size_t i = 0; i < _servantList.size(); i++)
- {
- ServantProxy* prx = _servantList[i].get();
- travelObjectProxys(prx, [&](ObjectProxy *op){
- objectProxys.push_back(op);
- });
- }
- travelObjectProxys(this, [&](ObjectProxy *op){
- objectProxys.push_back(op);
- });
- return objectProxys;
- }
- void ServantProxy::tars_update_endpoints(const set<EndpointInfo> &active, const set<EndpointInfo> &inactive)
- {
- onNotifyEndpoints(NULL, active, inactive);
- }
- void ServantProxy::onNotifyEndpoints(CommunicatorEpoll *communicatorEpoll, const set<EndpointInfo> &active, const set<EndpointInfo> &inactive)
- {
- //更新子servant proxy的地址
- for (size_t i = 0; i < _servantList.size(); i++)
- {
- ServantProxy* prx = _servantList[i].get();
- travelObjectProxys(prx, [&](ObjectProxy *op){
- if(op->getEndpointManager())
- {
- op->getEndpointManager()->updateEndpointsOutter(active, inactive);
- }
- });
- }
- travelObjectProxys(this, [&](ObjectProxy *op){
- if(op->getEndpointManager())
- {
- op->getEndpointManager()->updateEndpointsOutter(active, inactive);
- }
- });
- }
- void ServantProxy::onSetInactive(const EndpointInfo &ep)
- {
- if (!_rootPrx)
- return;
- for (size_t i = 0; i < _rootPrx->_servantList.size(); i++)
- {
- ServantPrx &prx = _rootPrx->_servantList[i];
- prx->forEachObject([&](ObjectProxy *o) { o->onSetInactive(ep); });
- }
- }
- int ServantProxy::servant_invoke(ReqMessage *msg, bool bCoroAsync)
- {
- ServantPrx prx = getServantPrx(msg);
- if (msg->callback)
- {
- msg->callback->setServantPrx(prx);
- }
- prx->invoke(msg, bCoroAsync);
- return 0;
- }
- void ServantProxy::http_call(const string &funcName, shared_ptr<TC_HttpRequest> &request, shared_ptr<TC_HttpResponse> &response)
- {
- if (_connectionSerial <= 0)
- {
- _connectionSerial = DEFAULT_CONNECTION_SERIAL;
- }
- ReqMessage *msg = new ReqMessage();
- msg->init(ReqMessage::SYNC_CALL, this);
- msg->bFromRpc = true;
- msg->request.sServantName = _objectProxy->name();
- msg->request.sFuncName = funcName;
- msg->request.sBuffer.resize(sizeof(shared_ptr<TC_HttpRequest>));
- msg->deconstructor = [msg] {
- shared_ptr<TC_HttpRequest> &data = *(shared_ptr<TC_HttpRequest> *)(msg->request.sBuffer.data());
- data.reset();
- if (!msg->response->sBuffer.empty())
- {
- shared_ptr<TC_HttpResponse> &rsp = *(shared_ptr<TC_HttpResponse> *)(msg->response->sBuffer.data());
- //主动reset一次
- rsp.reset();
- msg->response->sBuffer.clear();
- }
- };
- shared_ptr<TC_HttpRequest> &data = *(shared_ptr<TC_HttpRequest> *)(msg->request.sBuffer.data());
- data = request;
- servant_invoke(msg, false);
- response = *(shared_ptr<TC_HttpResponse> *)(msg->response->sBuffer.data());
- delete msg;
- msg = NULL;
- }
- void ServantProxy::http_call_async(const string &funcName, shared_ptr<TC_HttpRequest> &request, const HttpCallbackPtr &cb, bool bCoro)
- {
- if (_connectionSerial <= 0)
- {
- _connectionSerial = DEFAULT_CONNECTION_SERIAL;
- }
- ReqMessage *msg = new ReqMessage();
- msg->init(ReqMessage::ASYNC_CALL, this);
- msg->bFromRpc = true;
- msg->request.sServantName = _objectProxy->name();
- msg->request.sFuncName = funcName;
- msg->request.sBuffer.resize(sizeof(shared_ptr<TC_HttpRequest>));
- msg->deconstructor = [msg] {
- shared_ptr<TC_HttpRequest> &data = *(shared_ptr<TC_HttpRequest> *)(msg->request.sBuffer.data());
- data.reset();
- if (!msg->response->sBuffer.empty())
- {
- shared_ptr<TC_HttpResponse> &rsp = *(shared_ptr<TC_HttpResponse> *)(msg->response->sBuffer.data());
- //主动reset一次
- rsp.reset();
- msg->response->sBuffer.clear();
- }
- };
- *(shared_ptr<TC_HttpRequest> *)(msg->request.sBuffer.data()) = request;
- ServantProxyCallbackPtr callback = new HttpServantProxyCallback(cb);
- msg->callback = callback;
- servant_invoke(msg, bCoro);
- }
- //
- //选取一个网络线程对应的信息
- void ServantProxy::selectNetThreadInfo(ServantProxyThreadData *pSptd, ObjectProxy *&pObjProxy, shared_ptr<ReqInfoQueue> &pReqQ)
- {
- if(pSptd->_sched && pSptd->_communicatorEpoll == NULL)
- {
- //处于业务线程中, 且当前业务线程是以协程模式启动;
- auto schedCommunicatorEpollInfo = pSptd->getSchedCommunicatorEpollInfo(_communicator);
- shared_ptr<CommunicatorEpoll> ce;
- if (!schedCommunicatorEpollInfo->_communicator)
- {
- //当前协程没有关联过私有网络通信器, 需要新建!
- pReqQ = std::make_shared<ReqInfoQueue>(_communicator->getCommunicatorEpoll(0)->getNoSendQueueLimit());
- ce = _communicator->createSchedCommunicatorEpoll(pSptd->_reqQNo, pReqQ);
- schedCommunicatorEpollInfo->_communicator = _communicator;
- schedCommunicatorEpollInfo->_info._reqQueue = pReqQ;
- schedCommunicatorEpollInfo->_info._communicatorEpoll = ce;
- pObjProxy = ce->createObjectProxy(this, this->tars_full_name(), this->tars_setName());
- pObjProxy->initialize(true);
- }
- else
- {
- //网络通信器已经初始化过, 直接获取对象
- pReqQ = schedCommunicatorEpollInfo->_info._reqQueue.lock();
- ce = schedCommunicatorEpollInfo->_info._communicatorEpoll.lock();
- if(ce && pReqQ)
- {
- pObjProxy = ce->hasObjectProxy(this->tars_full_name(), this->tars_setName());
- //创建对应的ObjectProxy
- if(!pObjProxy)
- {
- pObjProxy = ce->createObjectProxy(this, this->tars_full_name(), this->tars_setName());
- pObjProxy->initialize(true);
- }
- }
- else
- {
- // assert(false);
- throw TarsCommunicatorException("communicator may deconstruct");
- }
- }
- }
- else
- {
- if( pSptd->_communicatorEpoll )
- {
- auto info = pSptd->getCommunicatorEpollInfo(_communicator);
- assert(info->_info.size() == 1);
- assert(info->_info[0]._communicatorEpoll.lock().get() == pSptd->_communicatorEpoll);
- pObjProxy = pSptd->_communicatorEpoll->servantToObjectProxy(this);
- pReqQ = info->_info[0]._reqQueue.lock();
- }
- else
- {
- //处于普通线程中, 判断当前线程是否关联过网络通信器, 没关联, 则关联所有的公有通信器, 即创建和公有网络通信器的队列!
- auto communicatorEpollInfo = pSptd->getCommunicatorEpollInfo(_communicator);
- //当前线程没有关联业务通信器, 需要关联
- if (!communicatorEpollInfo->_communicator)
- {
- communicatorEpollInfo->_communicator = _communicator;
- //为每个网络线程都创建一个队列
- for (size_t i = 0; i < _communicator->getCommunicatorEpollNum(); ++i)
- {
- shared_ptr<CommunicatorEpoll> ce = _communicator->getCommunicatorEpoll(i);
- pSptd->addCommunicatorEpoll(ce);
- }
- }
- assert(communicatorEpollInfo->_netSeq < _communicator->getCommunicatorEpollNum());
- //循环使用下一个网络线程发送数据
- auto ce = communicatorEpollInfo->_info[communicatorEpollInfo->_netSeq]._communicatorEpoll.lock();
- if (ce)
- {
- pObjProxy = ce->servantToObjectProxy(this);
- pReqQ = communicatorEpollInfo->_info[communicatorEpollInfo->_netSeq]._reqQueue.lock();
- communicatorEpollInfo->_netSeq++;
- if (communicatorEpollInfo->_netSeq == _communicator->getCommunicatorEpollNum())
- communicatorEpollInfo->_netSeq = 0;
- }
- }
- if(!pReqQ)
- {
- //队列已经析构, 说明通信器已经释放了!
- throw TarsCommunicatorException("communicator has deconstructed");
- }
- }
- }
- void ServantProxy::checkDye(RequestPacket& req)
- {
- //线程私有数据
- ServantProxyThreadData *pSptd = ServantProxyThreadData::getData();
- assert(pSptd != NULL);
- if (pSptd && pSptd->_data._dyeing)
- {
- SET_MSG_TYPE(req.iMessageType, TARSMESSAGETYPEDYED);
- req.status[ServantProxy::STATUS_DYED_KEY] = pSptd->_data._dyeingKey;
- }
- }
- void ServantProxy::checkTrace(RequestPacket &req)
- {
- //线程私有数据
- ServantProxyThreadData *pSptd = ServantProxyThreadData::getData();
- assert(pSptd != NULL);
- if (pSptd && pSptd->_traceCall)
- {
- SET_MSG_TYPE(req.iMessageType, tars::TARSMESSAGETYPETRACE);
- req.status[ServantProxy::STATUS_TRACE_KEY] = pSptd->getTraceKey();
- }
- }
- void ServantProxy::checkCookie(RequestPacket &req)
- {
- //线程私有数据
- ServantProxyThreadData *pSptd = ServantProxyThreadData::getData();
- assert(pSptd != NULL);
- std::for_each(pSptd->_data._cookie.begin(), pSptd->_data._cookie.end(), [&](map<string, string>::value_type &p) {
- req.status.insert(make_pair(p.first, p.second));
- });
- }
- void ServantProxy::tars_endpoints(vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint)
- {
- _endpointInfo->getEndpoint(activeEndPoint, inactiveEndPoint);
- }
- void ServantProxy::tars_endpointsAll(vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint)
- {
- _endpointInfo->getEndpointByAll(activeEndPoint, inactiveEndPoint);
- }
- void ServantProxy::tars_endpointsBySet(const string & sName,vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint)
- {
- _endpointInfo->getEndpointBySet(sName, activeEndPoint, inactiveEndPoint);
- }
- void ServantProxy::tars_endpointsByStation(const string & sName,vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint)
- {
- _endpointInfo->getEndpointByStation(sName, activeEndPoint, inactiveEndPoint);
- }
- vector<TC_Endpoint> ServantProxy::tars_endpoints(const std::string & sStation)
- {
- vector<TC_Endpoint> activeEndPoint;
- vector<TC_Endpoint> inactiveEndPoint;
- _endpointInfo->getTCEndpoint(activeEndPoint, inactiveEndPoint);
- return activeEndPoint;
- }
- void ServantProxy::tars_endpoints(const std::string & sStation, vector<TC_Endpoint> & vecActive, vector<TC_Endpoint> & vecInactive)
- {
- _endpointInfo->getTCEndpointByStation(sStation, vecActive, vecInactive);
- }
- vector<TC_Endpoint> ServantProxy::getEndpoint()
- {
- vector<TC_Endpoint> activeEndPoint;
- vector<TC_Endpoint> inactiveEndPoint;
- _endpointInfo->getTCEndpoint(activeEndPoint, inactiveEndPoint);
- return activeEndPoint;
- }
- vector<TC_Endpoint> ServantProxy::getEndpoint4All()
- {
- vector<TC_Endpoint> activeEndPoint;
- vector<TC_Endpoint> inactiveEndPoint;
- _endpointInfo->getTCEndpointByAll(activeEndPoint, inactiveEndPoint);
- return activeEndPoint;
- }
- //////////////////////////////////////////////////////////////////////////////////////////////////
- }
|