text: /**
  * Tencent is pleased to support the open source community by making DCache available.
  * Copyright (C) 2019 THL 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.
  */
  #ifndef __MKVCACHE_CALLBACK_H_
  #define __MKVCACHE_CALLBACK_H_
  
  #include "MKCache.h"
  #include "MKWCache.h"
  #include "CacheCallbackComm.h"
  #include "Global.h"
  
  using namespace std;
  using namespace tars;
  using namespace DCache;
  
  #define MKCacheCallbackLog(class_name, module_name, main_key, object_name, ret) TLOGDEBUG(LOG_MEM_FUN(class_name) \
                                                                                            << " mk:" << main_key << "|module:" << module_name << "|object:" << object_name << "|ret:" << ret << endl)
  
  #define MKCacheCallbackExcLog(class_name, module_name, main_key, object_name, ret) TLOGDEBUG(LOG_MEM_FUN(class_name) \
                                                                                               << " mk:" << main_key << "|module:" << module_name << "|object:" << object_name << "|errno:" << ret << endl)
  
  #define MKCacheCallbackDo(module_name, main_key, object_name, ret, rsp, recall_fun, rsp_fun) \
      MKCacheCallbackLog(MKCacheCallback, module_name, main_key, object_name, ret);            \
      ResponserPtr responser = make_responser(rsp_fun, rsp);                                   \
      procCallback(ret, recall_fun, responser)
  
  #define MKCacheCallbackExcDo(module_name, main_key, object_name, ret, rsp_para, rsp_fun) \
      MKCacheCallbackExcLog(MKCacheCallback, module_name, main_key, object_name, ret);     \
      rsp_para rsp;                                                                        \
      ResponserPtr responser = make_responser(rsp_fun, rsp);                               \
      procExceptionCall(ret, responser)
  
  #define MKWCacheCallbackDo(module_name, main_key, object_name, ret, recall_fun, rsp_fun) \
      MKCacheCallbackLog(MKWCacheCallback, module_name, main_key, object_name, ret);       \
      ResponserPtr responser = make_responser(rsp_fun);                                    \
      procCallback(main_key, ret, recall_fun, responser)
  
  #define MKWCacheCallbackExcDo(module_name, main_key, object_name, ret, rsp_fun)       \
      MKCacheCallbackExcLog(MKWCacheCallback, module_name, main_key, object_name, ret); \
      ResponserPtr responser = make_responser(rsp_fun);                                 \
      procExceptionCall(main_key, ret, responser)
  
  #define MKCacheBatchCallbackLog(class_name, module_name, object_name, ret) TLOGDEBUG(LOG_MEM_FUN(class_name) \
                                                                                       << " module:" << module_name << "|object:" << object_name << "|ret:" << ret << endl)
  
  #define MKCacheBatchCallbackExcLog(class_name, module_name, object_name, ret) TLOGDEBUG(LOG_MEM_FUN(class_name) \
                                                                                          << " module:" << module_name << "|object:" << object_name << "|errno:" << ret << endl)
  
  #define MKCacheBatchCallbackDo(ret, rsp, reroute_fun, recall_fun, rsp_fun) \
      procCallback(ret, rsp, reroute_fun, recall_fun, rsp_fun)
  
  #define MKCacheBatchCallbackExcDo(module_name, object_name, ret, rsp_fun)            \
      MKCacheBatchCallbackExcLog(MKCacheBatchCallback, module_name, object_name, ret); \
      procExceptionCall(ret, rsp_fun)
  
  #define MKWCacheBatchCallbackDo(ret, rsp, data_errcode, recall_fun, rsp_fun) \
      procCallback(ret, rsp, data_errcode, recall_fun, rsp_fun)
  
  #define MKWCacheBatchCallbackExcDo(module_name, object_name, ret, data_errcode, rsp_fun) \
      MKCacheBatchCallbackExcLog(MKWCacheBatchCallback, module_name, object_name, ret);    \
      procExceptionCall(ret, data_errcode, rsp_fun)
  
  template <typename Request, typename ClassName>
  struct ProcMKCacheCallback : public CacheCallbackComm
  {
  
      ProcMKCacheCallback(TarsCurrentPtr &current,
                          const Request &req,
                          const string &objectName,
                          const int64_t beginTime,
                          const string &idcArea = "",
                          const bool repeatFlag = false)
          : CacheCallbackComm(current, req.moduleName, objectName, beginTime, repeatFlag), _idcArea(idcArea), _req(req)
      {
      }
      virtual ~ProcMKCacheCallback() {}
  
    protected:
      /**
           * 对回调做统一处理
           * @param ret，接口返回值
           * @param recallmf，CacheServer接口函数指针
           * @param responser，负责应答的对象
           * @return void
           */
      void procCallback(const int ret, void (MKCacheProxy::*recallmf)(MKCachePrxCallbackPtr, const Request &, const map<string, string> &), const ResponserPtr responser)
      {
          const string &caller = _current->getContext()[CONTEXT_CALLER];
          const string &mainKey = _req.mainKey;
          const string &moduleName = _req.moduleName;
  
          if (ret != ET_SUCC)
          {
              FDLOG("CBError") << caller << "|mk:" << mainKey << "|module:" << moduleName << "|object:" << _objectName << "|ret:" << ret << endl;
  
              if ((ret == ET_KEY_AREA_ERR || ret == ET_SERVER_TYPE_ERR) && (!_repeatFlag))
              {
                  reportException("CBError");
  
                  string errMsg;
                  try
                  {
                      string objectName;
                      MKCachePrx prxCache;
  
                      // 根据模块名和mainkey重新获取要访问的cache服务的objectname以及代理，期间会强制更新路由
                      int ret = getCacheProxy(moduleName, mainKey, objectName, prxCache, _idcArea);
  
                      if (ret != ET_SUCC)
                      {
                          doResponse(ret, caller, EXCE, responser);
  
                          return;
                      }
  
                      MKCachePrxCallbackPtr cb = new ClassName(_current, _req, objectName, _beginTime, _idcArea, true);
                      //重新访问CacheServer
                      (((MKCacheProxy *)prxCache.get())->*recallmf)(cb, _req, map<string, string>());
  
                      return;
                  }
                  catch (exception &ex)
                  {
                      errMsg = string("exception:") + ex.what();
                  }
                  catch (...)
                  {
                      errMsg = "catch unkown exception";
                  }
  
                  TLOGERROR("MKCacheCallback::procCallback " << errMsg << endl);
  
                  FDLOG("CBError") << caller << "|mk:" << mainKey << "|module:" << moduleName << "|recall error:" << errMsg << endl;
  
                  reportException("CBError");
  
                  doResponse(ET_CACHE_ERR, caller, EXCE, responser);
  
                  return;
              }
          }
  
          doResponse(ret, caller, SUCC, responser);
      }
  
      /**
           * 对异常回调做统一处理
           * @param ret，接口返回值
           * @param responser，负责应答的对象
           * @return void
           */
      void procExceptionCall(const int ret, const ResponserPtr responser)
      {
          const string &caller = _current->getContext()[CONTEXT_CALLER];
  
          FDLOG("CBError") << caller << "|mk:" << _req.mainKey << "|module:" << _req.moduleName << "|object:" << _objectName << "|errno:" << ret << endl;
  
          reportException("CBError");
  
          if (ret == CALLTIMEOUT)
          {
              doResponse(ET_CACHE_ERR, caller, TIME_OUT, responser);
          }
          else
          {
              doResponse(ET_CACHE_ERR, caller, EXCE, responser);
          }
      }
  
    protected:
      string _idcArea; // 访问的模块存在镜像时，指定要访问的cache所属地区
      Request _req;
  };
  
  struct GetMKVCallback : public ProcMKCacheCallback<GetMKVReq, GetMKVCallback>, public MKCachePrxCallback
  {
      GetMKVCallback(TarsCurrentPtr &current,
                     const GetMKVReq &req,
                     const string &objectName,
                     const int64_t beginTime,
                     const string &idcArea = "",
                     const bool repeatFlag = false)
          : ProcMKCacheCallback<GetMKVReq, GetMKVCallback>(current, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetMKVCallback() {}
      virtual void callback_getMKV(int ret, const GetMKVRsp &rsp)
      {
          MKCacheCallbackDo(_req.moduleName, _req.mainKey, _objectName, ret, rsp, &MKCacheProxy::async_getMKV, &Proxy::async_response_getMKV);
      }
  
      virtual void callback_getMKV_exception(int ret)
      {
          MKCacheCallbackExcDo(_req.moduleName, _req.mainKey, _objectName, ret, GetMKVRsp, &Proxy::async_response_getMKV);
      }
  };
  
  struct GetMainKeyCountCallback : public ProcMKCacheCallback<MainKeyReq, GetMainKeyCountCallback>, public MKCachePrxCallback
  {
      GetMainKeyCountCallback(TarsCurrentPtr &current,
                              const MainKeyReq &req,
                              const string &objectName,
                              const int64_t beginTime,
                              const string &idcArea = "",
                              const bool repeatFlag = false)
          : ProcMKCacheCallback<MainKeyReq, GetMainKeyCountCallback>(current, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetMainKeyCountCallback() {}
      virtual void callback_getMainKeyCount(int ret)
      {
          MKCacheCallbackLog(MKCacheCallback, _req.moduleName, _req.mainKey, _objectName, ret);
  
          ResponserPtr responser = make_responser(&Proxy::async_response_getMainKeyCount);
  
          procCallback(ret, &MKCacheProxy::async_getMainKeyCount, responser);
      }
  
      virtual void callback_getMainKeyCount_exception(int ret)
      {
          MKCacheCallbackExcLog(MKCacheCallback, _req.moduleName, _req.mainKey, _objectName, ret);
  
          ResponserPtr responser = make_responser(&Proxy::async_response_getMainKeyCount);
  
          procExceptionCall(ret, responser);
      }
  };
  
  struct GetListCallback : public ProcMKCacheCallback<GetListReq, GetListCallback>, public MKCachePrxCallback
  {
      GetListCallback(TarsCurrentPtr &current,
                      const GetListReq &req,
                      const string &objectName,
                      const int64_t beginTime,
                      const string &idcArea = "",
                      const bool repeatFlag = false)
          : ProcMKCacheCallback<GetListReq, GetListCallback>(current, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetListCallback() {}
  
      virtual void callback_getList(int ret, const GetListRsp &rsp)
      {
          MKCacheCallbackDo(_req.moduleName, _req.mainKey, _objectName, ret, rsp, &MKCacheProxy::async_getList, &Proxy::async_response_getList);
      }
  
      virtual void callback_getList_exception(int ret)
      {
          MKCacheCallbackExcDo(_req.moduleName, _req.mainKey, _objectName, ret, GetListRsp, &Proxy::async_response_getList);
      }
  };
  
  struct GetRangeListCallback : public ProcMKCacheCallback<GetRangeListReq, GetRangeListCallback>, public MKCachePrxCallback
  {
      GetRangeListCallback(TarsCurrentPtr &current,
                           const GetRangeListReq &req,
                           const string &objectName,
                           const int64_t beginTime,
                           const string &idcArea = "",
                           const bool repeatFlag = false)
          : ProcMKCacheCallback<GetRangeListReq, GetRangeListCallback>(current, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetRangeListCallback() {}
  
      virtual void callback_getRangeList(int ret, const BatchEntry &rsp)
      {
          MKCacheCallbackDo(_req.moduleName, _req.mainKey, _objectName, ret, rsp, &MKCacheProxy::async_getRangeList, &Proxy::async_response_getRangeList);
      }
  
      virtual void callback_getRangeList_exception(int ret)
      {
          MKCacheCallbackExcDo(_req.moduleName, _req.mainKey, _objectName, ret, BatchEntry, &Proxy::async_response_getRangeList);
      }
  };
  
  struct GetSetCallback : public ProcMKCacheCallback<GetSetReq, GetSetCallback>, public MKCachePrxCallback
  {
      GetSetCallback(TarsCurrentPtr &current,
                     const GetSetReq &req,
                     const string &objectName,
                     const int64_t beginTime,
                     const string &idcArea = "",
                     const bool repeatFlag = false)
          : ProcMKCacheCallback<GetSetReq, GetSetCallback>(current, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetSetCallback() {}
  
      virtual void callback_getSet(int ret, const BatchEntry &rsp)
      {
          MKCacheCallbackDo(_req.moduleName, _req.mainKey, _objectName, ret, rsp, &MKCacheProxy::async_getSet, &Proxy::async_response_getSet);
      }
  
      virtual void callback_getSet_exception(int ret)
      {
          MKCacheCallbackExcDo(_req.moduleName, _req.mainKey, _objectName, ret, BatchEntry, &Proxy::async_response_getSet);
      }
  };
  
  struct GetZSetScoreCallback : public ProcMKCacheCallback<GetZsetScoreReq, GetZSetScoreCallback>, public MKCachePrxCallback
  {
      GetZSetScoreCallback(TarsCurrentPtr &current,
                           const GetZsetScoreReq &req,
                           const string &objectName,
                           const int64_t beginTime,
                           const string &idcArea = "",
                           const bool repeatFlag = false)
          : ProcMKCacheCallback<GetZsetScoreReq, GetZSetScoreCallback>(current, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetZSetScoreCallback() {}
  
      virtual void callback_getZSetScore(int ret, double score)
      {
          MKCacheCallbackDo(_req.moduleName, _req.mainKey, _objectName, ret, score, &MKCacheProxy::async_getZSetScore, &Proxy::async_response_getZSetScore);
      }
  
      virtual void callback_getZSetScore_exception(int ret)
      {
          MKCacheCallbackExcDo(_req.moduleName, _req.mainKey, _objectName, ret, double, &Proxy::async_response_getZSetScore);
      }
  };
  
  struct GetZSetPosCallback : public ProcMKCacheCallback<GetZsetPosReq, GetZSetPosCallback>, public MKCachePrxCallback
  {
      GetZSetPosCallback(TarsCurrentPtr &current,
                         const GetZsetPosReq &req,
                         const string &objectName,
                         const int64_t beginTime,
                         const string &idcArea = "",
                         const bool repeatFlag = false)
          : ProcMKCacheCallback<GetZsetPosReq, GetZSetPosCallback>(current, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetZSetPosCallback() {}
  
      virtual void callback_getZSetPos(int ret, tars::Int64 pos)
      {
          MKCacheCallbackDo(_req.moduleName, _req.mainKey, _objectName, ret, pos, &MKCacheProxy::async_getZSetPos, &Proxy::async_response_getZSetPos);
      }
  
      virtual void callback_getZSetPos_exception(int ret)
      {
          MKCacheCallbackExcDo(_req.moduleName, _req.mainKey, _objectName, ret, long long, &Proxy::async_response_getZSetPos);
      }
  };
  
  struct GetZSetByPosCallback : public ProcMKCacheCallback<GetZsetByPosReq, GetZSetByPosCallback>, public MKCachePrxCallback
  {
      GetZSetByPosCallback(TarsCurrentPtr &current,
                           const GetZsetByPosReq &req,
                           const string &objectName,
                           const int64_t beginTime,
                           const string &idcArea = "",
                           const bool repeatFlag = false)
          : ProcMKCacheCallback<GetZsetByPosReq, GetZSetByPosCallback>(current, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetZSetByPosCallback() {}
  
      virtual void callback_getZSetByPos(int ret, const BatchEntry &rsp)
      {
          MKCacheCallbackDo(_req.moduleName, _req.mainKey, _objectName, ret, rsp, &MKCacheProxy::async_getZSetByPos, &Proxy::async_response_getZSetByPos);
      }
  
      virtual void callback_getZSetByPos_exception(int ret)
      {
          MKCacheCallbackExcDo(_req.moduleName, _req.mainKey, _objectName, ret, BatchEntry, &Proxy::async_response_getZSetByPos);
      }
  };
  
  struct GetZSetByScoreCallback : public ProcMKCacheCallback<GetZsetByScoreReq, GetZSetByScoreCallback>, public MKCachePrxCallback
  {
      GetZSetByScoreCallback(TarsCurrentPtr &current,
                             const GetZsetByScoreReq &req,
                             const string &objectName,
                             const int64_t beginTime,
                             const string &idcArea = "",
                             const bool repeatFlag = false)
          : ProcMKCacheCallback<GetZsetByScoreReq, GetZSetByScoreCallback>(current, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetZSetByScoreCallback() {}
  
      virtual void callback_getZSetByScore(int ret, const BatchEntry &rsp)
      {
          MKCacheCallbackDo(_req.moduleName, _req.mainKey, _objectName, ret, rsp, &MKCacheProxy::async_getZSetByScore, &Proxy::async_response_getZSetByScore);
      }
  
      virtual void callback_getZSetByScore_exception(int ret)
      {
          MKCacheCallbackExcDo(_req.moduleName, _req.mainKey, _objectName, ret, BatchEntry, &Proxy::async_response_getZSetByScore);
      }
  };
  
  //////////////////////////////////////////////////////////////////////////
  
  template <typename Request, typename ClassName>
  struct ProcMKWCacheCallback : public CacheCallbackComm
  {
      ProcMKWCacheCallback(TarsCurrentPtr &current,
                           const Request &req,
                           const string &mainKey,
                           const string &objectName,
                           const int64_t beginTime,
                           const bool repeatFlag = false)
          : CacheCallbackComm(current, req.moduleName, objectName, beginTime, repeatFlag), _mainKey(mainKey), _req(req)
      {
      }
      virtual ~ProcMKWCacheCallback() {}
  
    protected:
      /**
           * 对回调做统一处理
           * @param ret，接口返回值
           * @param recallmf，CacheServer接口函数指针
           * @param responser，负责应答的对象
           * @return void
           */
      void procCallback(const string &mainKey, const int ret, void (MKWCacheProxy::*recallmf)(MKWCachePrxCallbackPtr, const Request &, const map<string, string> &), const ResponserPtr responser)
      {
          const string &caller = _current->getContext()[CONTEXT_CALLER];
          const string &moduleName = _req.moduleName;
  
          if (ret != ET_SUCC)
          {
              FDLOG("CBError") << caller << "|mk:" << mainKey << "|module:" << moduleName << "|object:" << _objectName << "|ret:" << ret << endl;
  
              if ((ret == ET_KEY_AREA_ERR || ret == ET_SERVER_TYPE_ERR) && (!_repeatFlag))
              {
                  reportException("CBError");
  
                  string errMsg;
                  try
                  {
                      string objectName;
                      MKWCachePrx prxCache;
  
                      // 根据模块名和key重新获取要访问的cache服务的objectname以及代理，期间会强制更新路由
                      int ret = getWCacheProxy(moduleName, mainKey, objectName, prxCache);
  
                      if (ret != ET_SUCC)
                      {
                          doResponse(ret, caller, EXCE, responser);
  
                          return;
                      }
  
                      MKWCachePrxCallbackPtr cb = new ClassName(_current, _req, objectName, _beginTime, true);
                      //重新访问CacheServer
                      (((MKWCacheProxy *)prxCache.get())->*recallmf)(cb, _req, map<string, string>());
  
                      return;
                  }
                  catch (exception &ex)
                  {
                      errMsg = string("exception:") + ex.what();
                  }
                  catch (...)
                  {
                      errMsg = "catch unkown exception";
                  }
  
                  TLOGERROR("MKWCacheCallback::procCallback " << errMsg << endl);
  
                  FDLOG("CBError") << caller << "|mk:" << mainKey << "|module:" << moduleName << "|recall error:" << errMsg << endl;
  
                  reportException("CBError");
  
                  doResponse(ET_SYS_ERR, caller, EXCE, responser);
  
                  return;
              }
          }
  
          doResponse(ret, caller, SUCC, responser);
      }
  
      /**
           * 对异常回调做统一处理
           * @param ret，接口返回值
           * @param responser，负责应答的对象
           * @return void
           */
      void procExceptionCall(const string &mainKey, const int ret, const ResponserPtr responser)
      {
          const string &caller = _current->getContext()[CONTEXT_CALLER];
  
          FDLOG("CBError") << caller << "|mk:" << mainKey << "|module:" << _req.moduleName << "|object:" << _objectName << "|errno:" << ret << endl;
  
          reportException("CBError");
  
          if (ret == CALLTIMEOUT)
          {
              doResponse(ET_CACHE_ERR, caller, TIME_OUT, responser);
          }
          else
          {
              doResponse(ET_CACHE_ERR, caller, EXCE, responser);
          }
      }
  
    protected:
      string _mainKey;
      Request _req;
  };
  
  struct InsertMKVCallback : public ProcMKWCacheCallback<InsertMKVReq, InsertMKVCallback>, public MKWCachePrxCallback
  {
      InsertMKVCallback(TarsCurrentPtr &current,
                        const InsertMKVReq &req,
                        const string &objectName,
                        const int64_t beginTime,
                        const bool repeatFlag = false)
          : ProcMKWCacheCallback<InsertMKVReq, InsertMKVCallback>(current, req, req.data.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~InsertMKVCallback() {}
  
      virtual void callback_insertMKV(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_insertMKV, &Proxy::async_response_insertMKV);
      }
  
      virtual void callback_insertMKV_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_insertMKV);
      }
  };
  
  struct UpdateMKVCallback : public ProcMKWCacheCallback<UpdateMKVReq, UpdateMKVCallback>, public MKWCachePrxCallback
  {
      UpdateMKVCallback(TarsCurrentPtr &current,
                        const UpdateMKVReq &req,
                        const string &objectName,
                        const int64_t beginTime,
                        const bool repeatFlag = false)
          : ProcMKWCacheCallback<UpdateMKVReq, UpdateMKVCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~UpdateMKVCallback() {}
  
      virtual void callback_updateMKV(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_updateMKV, &Proxy::async_response_updateMKV);
      }
  
      virtual void callback_updateMKV_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_updateMKV);
      }
  };
  
  struct UpdateMKVAtomCallback : public ProcMKWCacheCallback<UpdateMKVAtomReq, UpdateMKVAtomCallback>, public MKWCachePrxCallback
  {
      UpdateMKVAtomCallback(TarsCurrentPtr &current,
                            const UpdateMKVAtomReq &req,
                            const string &objectName,
                            const int64_t beginTime,
                            const bool repeatFlag = false)
          : ProcMKWCacheCallback<UpdateMKVAtomReq, UpdateMKVAtomCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~UpdateMKVAtomCallback() {}
  
      virtual void callback_updateMKVAtom(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_updateMKVAtom, &Proxy::async_response_updateMKVAtom);
      }
  
      virtual void callback_updateMKVAtom_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_updateMKVAtom);
      }
  };
  
  struct EraseMKVCallback : public ProcMKWCacheCallback<MainKeyReq, EraseMKVCallback>, public MKWCachePrxCallback
  {
      EraseMKVCallback(TarsCurrentPtr &current,
                       const MainKeyReq &req,
                       const string &objectName,
                       const int64_t beginTime,
                       const bool repeatFlag = false)
          : ProcMKWCacheCallback<MainKeyReq, EraseMKVCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~EraseMKVCallback() {}
  
      virtual void callback_eraseMKV(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_eraseMKV, &Proxy::async_response_eraseMKV);
      }
  
      virtual void callback_eraseMKV_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_eraseMKV);
      }
  };
  
  struct DelMKVCallback : public ProcMKWCacheCallback<DelMKVReq, DelMKVCallback>, public MKWCachePrxCallback
  {
      DelMKVCallback(TarsCurrentPtr &current,
                     const DelMKVReq &req,
                     const string &objectName,
                     const int64_t beginTime,
                     const bool repeatFlag = false)
          : ProcMKWCacheCallback<DelMKVReq, DelMKVCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~DelMKVCallback() {}
  
      virtual void callback_delMKV(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_delMKV, &Proxy::async_response_delMKV);
      }
  
      virtual void callback_delMKV_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_delMKV);
      }
  };
  
  struct PushListCallback : public ProcMKWCacheCallback<PushListReq, PushListCallback>, public MKWCachePrxCallback
  {
      PushListCallback(TarsCurrentPtr &current,
                       const PushListReq &req,
                       const string &objectName,
                       const int64_t beginTime,
                       const bool repeatFlag = false)
          : ProcMKWCacheCallback<PushListReq, PushListCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~PushListCallback() {}
      virtual void callback_pushList(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_pushList, &Proxy::async_response_pushList);
      }
  
      virtual void callback_pushList_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_pushList);
      }
  };
  
  struct PopListCallback : public ProcMKWCacheCallback<PopListReq, PopListCallback>, public MKWCachePrxCallback
  {
      PopListCallback(TarsCurrentPtr &current,
                      const PopListReq &req,
                      const string &objectName,
                      const int64_t beginTime,
                      const bool repeatFlag = false)
          : ProcMKWCacheCallback<PopListReq, PopListCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~PopListCallback() {}
      virtual void callback_popList(int ret, const PopListRsp &rsp)
      {
          MKCacheCallbackLog(MKWCacheCallback, _req.moduleName, _mainKey, _objectName, ret);
  
          ResponserPtr responser = make_responser(&Proxy::async_response_popList, rsp);
  
          procCallback(_mainKey, ret, &MKWCacheProxy::async_popList, responser);
      }
  
      virtual void callback_popList_exception(int ret)
      {
          MKCacheCallbackExcLog(MKWCacheCallback, _req.moduleName, _mainKey, _objectName, ret);
  
          PopListRsp rsp;
          ResponserPtr responser = make_responser(&Proxy::async_response_popList, rsp);
  
          procExceptionCall(_mainKey, ret, responser);
      }
  };
  
  struct ReplaceListCallback : public ProcMKWCacheCallback<ReplaceListReq, ReplaceListCallback>, public MKWCachePrxCallback
  {
      ReplaceListCallback(TarsCurrentPtr &current,
                          const ReplaceListReq &req,
                          const string &objectName,
                          const int64_t beginTime,
                          const bool repeatFlag = false)
          : ProcMKWCacheCallback<ReplaceListReq, ReplaceListCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~ReplaceListCallback() {}
  
      virtual void callback_replaceList(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_replaceList, &Proxy::async_response_replaceList);
      }
  
      virtual void callback_replaceList_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_replaceList);
      }
  };
  
  struct TrimListCallback : public ProcMKWCacheCallback<TrimListReq, TrimListCallback>, public MKWCachePrxCallback
  {
      TrimListCallback(TarsCurrentPtr &current,
                       const TrimListReq &req,
                       const string &objectName,
                       const int64_t beginTime,
                       const bool repeatFlag = false)
          : ProcMKWCacheCallback<TrimListReq, TrimListCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~TrimListCallback() {}
  
      virtual void callback_trimList(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_trimList, &Proxy::async_response_trimList);
      }
  
      virtual void callback_trimList_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_trimList);
      }
  };
  
  struct RemListCallback : public ProcMKWCacheCallback<RemListReq, RemListCallback>, public MKWCachePrxCallback
  {
      RemListCallback(TarsCurrentPtr &current,
                      const RemListReq &req,
                      const string &objectName,
                      const int64_t beginTime,
                      const bool repeatFlag = false)
          : ProcMKWCacheCallback<RemListReq, RemListCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~RemListCallback() {}
  
      virtual void callback_remList(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_remList, &Proxy::async_response_remList);
      }
  
      virtual void callback_remList_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_remList);
      }
  };
  
  struct AddSetCallback : public ProcMKWCacheCallback<AddSetReq, AddSetCallback>, public MKWCachePrxCallback
  {
      AddSetCallback(TarsCurrentPtr &current,
                     const AddSetReq &req,
                     const string &objectName,
                     const int64_t beginTime,
                     const bool repeatFlag = false)
          : ProcMKWCacheCallback<AddSetReq, AddSetCallback>(current, req, req.value.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~AddSetCallback() {}
  
      virtual void callback_addSet(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_addSet, &Proxy::async_response_addSet);
      }
  
      virtual void callback_addSet_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_addSet);
      }
  };
  
  struct DelSetCallback : public ProcMKWCacheCallback<DelSetReq, DelSetCallback>, public MKWCachePrxCallback
  {
      DelSetCallback(TarsCurrentPtr &current,
                     const DelSetReq &req,
                     const string &objectName,
                     const int64_t beginTime,
                     const bool repeatFlag = false)
          : ProcMKWCacheCallback<DelSetReq, DelSetCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~DelSetCallback() {}
  
      virtual void callback_delSet(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_delSet, &Proxy::async_response_delSet);
      }
  
      virtual void callback_delSet_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_delSet);
      }
  };
  
  struct AddZSetCallback : public ProcMKWCacheCallback<AddZSetReq, AddZSetCallback>, public MKWCachePrxCallback
  {
      AddZSetCallback(TarsCurrentPtr &current,
                      const AddZSetReq &req,
                      const string &objectName,
                      const int64_t beginTime,
                      const bool repeatFlag = false)
          : ProcMKWCacheCallback<AddZSetReq, AddZSetCallback>(current, req, req.value.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~AddZSetCallback() {}
  
      virtual void callback_addZSet(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_addZSet, &Proxy::async_response_addZSet);
      }
  
      virtual void callback_addZSet_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_addZSet);
      }
  };
  
  struct IncScoreZSetCallback : public ProcMKWCacheCallback<IncZSetScoreReq, IncScoreZSetCallback>, public MKWCachePrxCallback
  {
      IncScoreZSetCallback(TarsCurrentPtr &current,
                           const IncZSetScoreReq &req,
                           const string &objectName,
                           const int64_t beginTime,
                           const bool repeatFlag = false)
          : ProcMKWCacheCallback<IncZSetScoreReq, IncScoreZSetCallback>(current, req, req.value.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~IncScoreZSetCallback() {}
  
      virtual void callback_incScoreZSet(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_incScoreZSet, &Proxy::async_response_incScoreZSet);
      }
  
      virtual void callback_incScoreZSet_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_incScoreZSet);
      }
  };
  
  struct DelZSetCallback : public ProcMKWCacheCallback<DelZSetReq, DelZSetCallback>, public MKWCachePrxCallback
  {
      DelZSetCallback(TarsCurrentPtr &current,
                      const DelZSetReq &req,
                      const string &objectName,
                      const int64_t beginTime,
                      const bool repeatFlag = false)
          : ProcMKWCacheCallback<DelZSetReq, DelZSetCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~DelZSetCallback() {}
  
      virtual void callback_delZSet(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_delZSet, &Proxy::async_response_delZSet);
      }
  
      virtual void callback_delZSet_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_delZSet);
      }
  };
  
  struct DelZSetByScoreCallback : public ProcMKWCacheCallback<DelZSetByScoreReq, DelZSetByScoreCallback>, public MKWCachePrxCallback
  {
      DelZSetByScoreCallback(TarsCurrentPtr &current,
                             const DelZSetByScoreReq &req,
                             const string &objectName,
                             const int64_t beginTime,
                             const bool repeatFlag = false)
          : ProcMKWCacheCallback<DelZSetByScoreReq, DelZSetByScoreCallback>(current, req, req.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~DelZSetByScoreCallback() {}
  
      virtual void callback_delZSetByScore(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_delZSetByScore, &Proxy::async_response_delZSetByScore);
      }
  
      virtual void callback_delZSetByScore_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_delZSetByScore);
      }
  };
  
  struct UpdateZSetCallback : public ProcMKWCacheCallback<UpdateZSetReq, UpdateZSetCallback>, public MKWCachePrxCallback
  {
      UpdateZSetCallback(TarsCurrentPtr &current,
                         const UpdateZSetReq &req,
                         const string &objectName,
                         const int64_t beginTime,
                         const bool repeatFlag = false)
          : ProcMKWCacheCallback<UpdateZSetReq, UpdateZSetCallback>(current, req, req.value.mainKey, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~UpdateZSetCallback() {}
  
      virtual void callback_updateZSet(int ret)
      {
          MKWCacheCallbackDo(_req.moduleName, _mainKey, _objectName, ret, &MKWCacheProxy::async_updateZSet, &Proxy::async_response_updateZSet);
      }
  
      virtual void callback_updateZSet_exception(int ret)
      {
          MKWCacheCallbackExcDo(_req.moduleName, _mainKey, _objectName, ret, &Proxy::async_response_updateZSet);
      }
  };
  
  //////////////////////////////////////////////////////////////////////////
  
  template <typename Response>
  struct MKCacheBatchCallParam : public BatchCallParamComm
  {
      MKCacheBatchCallParam(const size_t count) : BatchCallParamComm(count), _end(false) {}
  
      bool setEnd()
      {
          TC_ThreadLock::Lock lock(_lock);
          if (_end)
          {
              return false;
          }
          else
          {
              _end = true;
          }
  
          return true;
      }
  
      bool _end;
  
      Response _rsp;
  };
  
  template <typename Request, typename Response, typename ClassName>
  struct ProcMKCacheBatchCallback : public CacheCallbackComm
  {
  
      ProcMKCacheBatchCallback(TarsCurrentPtr &current,
                               BatchCallParamPtr &param,
                               const Request &req,
                               const string &objectName,
                               const int64_t beginTime,
                               const string &idcArea = "",
                               const bool repeatFlag = false)
          : CacheCallbackComm(current, req.moduleName, objectName, beginTime, repeatFlag), _idcArea(idcArea), _param(param), _req(req)
      {
      }
      virtual ~ProcMKCacheBatchCallback() {}
  
    protected:
      /**
           * 对回调做统一处理
           * @param ret，接口返回值
           * @param recallmf，CacheServer接口函数指针
           * @param responser，负责应答的对象
           * @return void
           */
      void procCallback(int ret,
                        const Response &rsp,
                        int (ProcMKCacheBatchCallback::*reroutemf)(const RouterTable &, map<string, Request> &),
                        void (MKCacheProxy::*recallmf)(MKCachePrxCallbackPtr, const Request &, const map<string, string> &),
                        void (*resmf)(TarsCurrentPtr, int, const Response &))
      {
          MKCacheBatchCallParam<Response> *param = (MKCacheBatchCallParam<Response> *)(_param.get());
  
          if (param->_end)
          {
              return;
          }
  
          const string &caller = _current->getContext()[CONTEXT_CALLER];
          const string &moduleName = _req.moduleName;
  
          if (ret == ET_SUCC)
          {
              {
                  TC_ThreadLock::Lock lock(param->_lock);
  
                  param->_rsp.data.insert(param->_rsp.data.end(), rsp.data.begin(), rsp.data.end());
              }
  
              if (param->_count.dec() <= 0)
              {
                  ResponserPtr responser = make_responser(resmf, param->_rsp);
  
                  doResponse(ret, caller, SUCC, responser);
  
                  param->_end = true;
              }
          }
          else
          {
              FDLOG("CBError") << caller << "|module:" << moduleName << "|object:" << _objectName << "|ret:" << ret << endl;
  
              reportException("CBError");
  
              if (ret == ET_KEY_AREA_ERR || ret == ET_SERVER_TYPE_ERR)
              {
                  if (!_repeatFlag)
                  {
                      ret = doRecall(reroutemf, recallmf);
  
                      if (ret == ET_SUCC)
                      {
                          return;
                      }
                  }
                  else
                  {
                      ret = ET_SYS_ERR;
                  }
              }
              else
              {
                  // 将cache端返回的ET_SYS_ERR映射为ET_CACHE_ERR返回给客户端
                  if (ret == ET_SYS_ERR)
                  {
                      ret = ET_CACHE_ERR;
                  }
              }
  
              if (param->setEnd())
              {
                  Response rsp;
                  ResponserPtr responser = make_responser(resmf, rsp);
  
                  doResponse(ret, caller, SUCC, responser);
              }
          }
      }
  
      /**
           * 对异常回调做统一处理
           * @param ret，接口返回值
           * @param responser，负责应答的对象
           * @return void
           */
      void procExceptionCall(const int ret, void (*resmf)(TarsCurrentPtr, int, const Response &))
      {
          const string &caller = _current->getContext()[CONTEXT_CALLER];
  
          FDLOG("CBError") << caller << "|module:" << _req.moduleName << "|object:" << _objectName << "|errno:" << ret << endl;
  
          reportException("CBError");
  
          MKCacheBatchCallParam<Response> *param = (MKCacheBatchCallParam<Response> *)(_param.get());
          if (param->setEnd())
          {
              Response rsp;
              ResponserPtr responser = make_responser(resmf, rsp);
  
              if (ret == CALLTIMEOUT)
              {
                  doResponse(ET_CACHE_ERR, caller, TIME_OUT, responser);
              }
              else
              {
                  doResponse(ET_CACHE_ERR, caller, EXCE, responser);
              }
          }
      }
  
      /**
           * 对路由失败的key重试
           */
      int doRecall(int (ProcMKCacheBatchCallback::*reroutemf)(const RouterTable &, map<string, Request> &),
                   void (MKCacheProxy::*recallmf)(MKCachePrxCallbackPtr, const Request &, const map<string, string> &))
      {
          const string &moduleName = _req.moduleName;
  
          RouterTableInfo *routeTableInfo = g_app._routerTableInfoFactory->getRouterTableInfo(moduleName);
          if (!routeTableInfo)
          {
              TLOGERROR("MKCacheBatchCallback::doRecall do not support module:" << moduleName << endl);
  
              return ET_MODULE_NAME_INVALID;
          }
  
          int ret = routeTableInfo->update();
          if (ret != ET_SUCC)
          {
              TLOGERROR("MKCacheBatchCallback::doRecall update route table failed, module:" << moduleName << "|ret:" << ret << endl);
          }
          else
          {
              RouterTable &routeTable = routeTableInfo->getRouterTable();
  
              map<string, Request> objectReq;
  
              ret = (this->*reroutemf)(routeTable, objectReq);
  
              if (ret == ET_SUCC)
              {
                  _param->_count.add(objectReq.size() - 1);
  
                  typename map<string, Request>::const_iterator it = objectReq.begin();
                  typename map<string, Request>::const_iterator itEnd = objectReq.end();
                  for (; it != itEnd; ++it)
                  {
                      try
                      {
                          MKCachePrxCallbackPtr cb = new ClassName(_current, _param, it->second, it->first, _beginTime, _idcArea, true);
                          MKCachePrx prxCache;
                          ret = getProxy(it->first, prxCache);
                          if (ret == ET_SUCC)
                          {
                              //重新访问CacheServer
                              (((MKCacheProxy *)prxCache.get())->*recallmf)(cb, it->second, map<string, string>());
  
                              continue;
                          }
                          else
                          {
                              return ret;
                          }
                      }
                      catch (exception &ex)
                      {
                          TLOGERROR("MKCacheBatchCallback::doRecall exception:" << ex.what() << endl);
                      }
                      catch (...)
                      {
                          TLOGERROR("MKCacheBatchCallback::doRecall catch unkown exception" << endl);
                      }
  
                      return ET_CACHE_ERR;
                  }
              }
          }
  
          if (ret != ET_SUCC)
          {
              ret = ET_SYS_ERR; // 将各种错误统一转换为ET_SYS_ERR返回给客户端
          }
  
          return ret;
      }
  
    protected:
      int reRouteGetMKVBatch(const RouterTable &routeTable, map<string, MKVBatchReq> &objectReq)
      {
          int ret = ET_SUCC;
  
          for (size_t i = 0; i < _req.mainKeys.size(); ++i)
          {
              // 从路由表中获得要访问的CacheServer的节点信息
              ServerInfo serverInfo;
              if (_idcArea.empty())
              {
                  ret = routeTable.getMaster(_req.mainKeys[i], serverInfo);
              }
              else
              {
                  ret = routeTable.getIdcServer(_req.mainKeys[i], _idcArea, false, serverInfo);
              }
              if (ret != RouterTable::RET_SUCC)
              {
                  // 如果通过key无法获取CacheServer的节点信息
                  TLOGERROR("MKCacheBatchCallback::reRouteGetMKVBatch can not locate key:" << _req.mainKeys[i] << "|ret:" << ret << endl);
  
                  break;
              }
  
              const string &objectName = serverInfo.CacheServant;
              map<string, MKVBatchReq>::iterator it = objectReq.find(objectName);
              if (it == objectReq.end())
              {
                  MKVBatchReq req(_req);
                  req.mainKeys.clear();
                  pair<map<string, MKVBatchReq>::iterator, bool> p = objectReq.insert(make_pair(objectName, req));
                  it = p.first;
              }
              it->second.mainKeys.push_back(_req.mainKeys[i]);
          }
  
          return ret;
      }
  
      int reRouteGetMUKBatch(const RouterTable &routeTable, map<string, MUKBatchReq> &objectReq)
      {
          int ret = ET_SUCC;
  
          for (size_t i = 0; i < _req.primaryKeys.size(); ++i)
          {
              const string &mainKey = _req.primaryKeys[i].mainKey;
  
              // 从路由表中获得要访问的CacheServer的节点信息
              ServerInfo serverInfo;
              if (_idcArea.empty())
              {
                  ret = routeTable.getMaster(mainKey, serverInfo);
              }
              else
              {
                  ret = routeTable.getIdcServer(mainKey, _idcArea, false, serverInfo);
              }
              if (ret != RouterTable::RET_SUCC)
              {
                  // 如果通过key无法获取CacheServer的节点信息
                  TLOGERROR("MKCacheBatchCallback::reRouteGetMUKBatch can not locate key:" << mainKey << "|ret:" << ret << endl);
  
                  break;
              }
  
              const string &objectName = serverInfo.CacheServant;
              map<string, MUKBatchReq>::iterator it = objectReq.find(objectName);
              if (it == objectReq.end())
              {
                  MUKBatchReq req(_req);
                  req.primaryKeys.clear();
                  pair<map<string, MUKBatchReq>::iterator, bool> p = objectReq.insert(make_pair(objectName, req));
                  it = p.first;
              }
              it->second.primaryKeys.push_back(_req.primaryKeys[i]);
          }
  
          return ret;
      }
  
      int reRouteGetMKVBatchEx(const RouterTable &routeTable, map<string, MKVBatchExReq> &objectReq)
      {
          int ret = ET_SUCC;
  
          for (size_t i = 0; i < _req.cond.size(); ++i)
          {
              const string &mainKey = _req.cond[i].mainKey;
  
              // 从路由表中获得要访问的CacheServer的节点信息
              ServerInfo serverInfo;
              if (_idcArea.empty())
              {
                  ret = routeTable.getMaster(mainKey, serverInfo);
              }
              else
              {
                  ret = routeTable.getIdcServer(mainKey, _idcArea, false, serverInfo);
              }
              if (ret != RouterTable::RET_SUCC)
              {
                  // 如果通过key无法获取CacheServer的节点信息
                  TLOGERROR("MKCacheBatchCallback::reRouteGetMKVBatchEx can not locate key:" << mainKey << "|ret:" << ret << endl);
  
                  break;
              }
  
              const string &objectName = serverInfo.CacheServant;
              map<string, MKVBatchExReq>::iterator it = objectReq.find(objectName);
              if (it == objectReq.end())
              {
                  MKVBatchExReq req(_req);
                  req.cond.clear();
                  pair<map<string, MKVBatchExReq>::iterator, bool> p = objectReq.insert(make_pair(objectName, req));
                  it = p.first;
              }
              it->second.cond.push_back(_req.cond[i]);
          }
  
          return ret;
      }
  
    protected:
      string _idcArea;
      BatchCallParamPtr _param;
      Request _req;
  };
  
  struct GetMKVBatchCallback : public ProcMKCacheBatchCallback<MKVBatchReq, MKVBatchRsp, GetMKVBatchCallback>, public MKCachePrxCallback
  {
  
      GetMKVBatchCallback(TarsCurrentPtr &current,
                          BatchCallParamPtr &param,
                          const MKVBatchReq &req,
                          const string &objectName,
                          const int64_t beginTime,
                          const string &idcArea = "",
                          const bool repeatFlag = false)
          : ProcMKCacheBatchCallback<MKVBatchReq, MKVBatchRsp, GetMKVBatchCallback>(current, param, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetMKVBatchCallback() {}
  
      virtual void callback_getMKVBatch(int ret, const MKVBatchRsp &rsp)
      {
          MKCacheBatchCallbackDo(ret, rsp, &GetMKVBatchCallback::reRouteGetMKVBatch, &MKCacheProxy::async_getMKVBatch, &Proxy::async_response_getMKVBatch);
      }
  
      virtual void callback_getMKVBatch_exception(int ret)
      {
          MKCacheBatchCallbackExcDo(_req.moduleName, _objectName, ret, &Proxy::async_response_getMKVBatch);
      }
  };
  
  struct GetMUKBatchCallback : public ProcMKCacheBatchCallback<MUKBatchReq, MUKBatchRsp, GetMUKBatchCallback>, public MKCachePrxCallback
  {
  
      GetMUKBatchCallback(TarsCurrentPtr &current,
                          BatchCallParamPtr &param,
                          const MUKBatchReq &req,
                          const string &objectName,
                          const int64_t beginTime,
                          const string &idcArea = "",
                          const bool repeatFlag = false)
          : ProcMKCacheBatchCallback<MUKBatchReq, MUKBatchRsp, GetMUKBatchCallback>(current, param, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetMUKBatchCallback() {}
  
      virtual void callback_getMUKBatch(int ret, const MUKBatchRsp &rsp)
      {
          MKCacheBatchCallbackDo(ret, rsp, &GetMUKBatchCallback::reRouteGetMUKBatch, &MKCacheProxy::async_getMUKBatch, &Proxy::async_response_getMUKBatch);
      }
  
      virtual void callback_getMUKBatch_exception(int ret)
      {
          MKCacheBatchCallbackExcDo(_req.moduleName, _objectName, ret, &Proxy::async_response_getMUKBatch);
      }
  };
  
  struct GetMKVBatchExCallback : public ProcMKCacheBatchCallback<MKVBatchExReq, MKVBatchExRsp, GetMKVBatchExCallback>, public MKCachePrxCallback
  {
  
      GetMKVBatchExCallback(TarsCurrentPtr &current,
                            BatchCallParamPtr &param,
                            const MKVBatchExReq &req,
                            const string &objectName,
                            const int64_t beginTime,
                            const string &idcArea = "",
                            const bool repeatFlag = false)
          : ProcMKCacheBatchCallback<MKVBatchExReq, MKVBatchExRsp, GetMKVBatchExCallback>(current, param, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetMKVBatchExCallback() {}
  
      virtual void callback_getMKVBatchEx(int ret, const MKVBatchExRsp &rsp)
      {
          MKCacheBatchCallbackDo(ret, rsp, &GetMKVBatchExCallback::reRouteGetMKVBatchEx, &MKCacheProxy::async_getMKVBatchEx, &Proxy::async_response_getMKVBatchEx);
      }
  
      virtual void callback_getMKVBatchEx_exception(int ret)
      {
          MKCacheBatchCallbackExcDo(_req.moduleName, _objectName, ret, &Proxy::async_response_getMKVBatchEx);
      }
  };
  
  struct GetAllMainKeyCallback : public ProcMKCacheBatchCallback<GetAllKeysReq, GetAllKeysRsp, GetAllMainKeyCallback>, public MKCachePrxCallback
  {
  
      GetAllMainKeyCallback(TarsCurrentPtr &current,
                            BatchCallParamPtr &param,
                            const GetAllKeysReq &req,
                            const string &objectName,
                            const int64_t beginTime,
                            const string &idcArea = "",
                            const bool repeatFlag = false)
          : ProcMKCacheBatchCallback<GetAllKeysReq, GetAllKeysRsp, GetAllMainKeyCallback>(current, param, req, objectName, beginTime, idcArea, repeatFlag)
      {
      }
      virtual ~GetAllMainKeyCallback() {}
  
      virtual void callback_getAllMainKey(int ret, const GetAllKeysRsp &rsp)
      {
          MKCacheBatchCallParam<GetAllKeysRsp> *param = (MKCacheBatchCallParam<GetAllKeysRsp> *)(_param.get());
  
          if (param->_end)
          {
              return;
          }
  
          if (ret == ET_SUCC)
          {
              {
                  TC_ThreadLock::Lock lock(param->_lock);
                  param->_rsp.keys.insert(param->_rsp.keys.end(), rsp.keys.begin(), rsp.keys.end());
              }
  
              if (!rsp.isEnd)
              {
                  param->_rsp.isEnd = false;
              }
  
              if (param->_count.dec() <= 0)
              {
                  ResponserPtr responser = make_responser(&Proxy::async_response_getAllMainKey, param->_rsp);
                  const string &caller = _current->getContext()[CONTEXT_CALLER];
                  doResponse(ret, caller, SUCC, responser);
  
                  param->_end = true;
              }
          }
          else
          {
              const string &caller = _current->getContext()[CONTEXT_CALLER];
              const string &moduleName = _req.moduleName;
  
              FDLOG("CBError") << caller << "|module:" << moduleName << "|object:" << _objectName << "|ret:" << ret << endl;
  
              reportException("CBError");
  
              if (param->setEnd())
              {
                  ResponserPtr responser = make_responser(&Proxy::async_response_getAllMainKey, rsp);
  
                  ret = (ret == ET_KEY_AREA_ERR || ret == ET_SERVER_TYPE_ERR) ? ET_SYS_ERR : ET_CACHE_ERR;
  
                  doResponse(ret, caller, SUCC, responser);
              }
          }
      }
  
      virtual void callback_getAllMainKey_exception(int ret)
      {
          MKCacheBatchCallbackExcDo(_req.moduleName, _objectName, ret, &Proxy::async_response_getAllMainKey);
      }
  };
  
  //////////////////////////////////////////////////////////////////////////
  
  template <typename Request>
  struct MKWCacheBatchCallParam : public BatchCallParamComm
  {
      MKWCacheBatchCallParam(const Request &req, const size_t count) : BatchCallParamComm(count)
      {
          _req = req;
      }
  
      void addResult(const map<int, int> &rspData)
      {
          TC_ThreadLock::Lock lock(_lock);
  
          _rsp.rspData.insert(rspData.begin(), rspData.end());
      }
  
      Request _req;
  
      MKVBatchWriteRsp _rsp;
  };
  
  template <typename Request, typename ClassName>
  struct ProcMKWCacheBatchCallback : public CacheCallbackComm
  {
      /**
       * 构造函数
       */
      ProcMKWCacheBatchCallback(TarsCurrentPtr &current,
                                BatchCallParamPtr &param,
                                const vector<size_t> &indexs,
                                const Request &req,
                                const string &objectName,
                                const int64_t beginTime,
                                const bool repeatFlag = false)
          : CacheCallbackComm(current, req.moduleName, objectName, beginTime, repeatFlag), _param(param), _indexs(indexs), _req(req)
      {
      }
      virtual ~ProcMKWCacheBatchCallback() {}
  
    protected:
      /**
           * 对回调做统一处理
           * @param ret，接口返回值
           * @param recallmf，CacheServer接口函数指针
           * @param responser，负责应答的对象
           * @return void
           */
      void procCallback(int ret, const MKVBatchWriteRsp &rsp, const int errcode, void (MKWCacheProxy::*recallmf)(MKWCachePrxCallbackPtr, const Request &, const map<string, string> &), void (*resmf)(TarsCurrentPtr, int, const MKVBatchWriteRsp &))
      {
          const string &caller = _current->getContext()[CONTEXT_CALLER];
  
          MKWCacheBatchCallParam<Request> *param = (MKWCacheBatchCallParam<Request> *)_param.get();
  
          if (ret == ET_SUCC)
          {
              if (!rsp.rspData.empty())
              {
                  map<int, int> rspData;
  
                  map<int, int>::const_reverse_iterator it = rsp.rspData.rbegin(), itEnd = rsp.rspData.rend();
  
                  assert(it->first < _indexs.size());
  
                  for (; it != itEnd; ++it)
                  {
                      rspData[_indexs[it->first]] = it->second;
                  }
  
                  param->addResult(rspData);
              }
          }
          else
          {
              const string &moduleName = _req.moduleName;
  
              FDLOG("CBError") << caller << "|module:" << moduleName << "|object:" << _objectName << "|ret:" << ret << endl;
  
              map<int, int> failedIndexRet;
  
              if (ret == ET_PARTIAL_FAIL)
              {
                  map<int, int>::const_reverse_iterator it = rsp.rspData.rbegin(), itEnd = rsp.rspData.rend();
  
                  assert((!rsp.rspData.empty()) && (it->first < _indexs.size()));
  
                  vector<size_t> repeatIndexs;
                  for (; it != itEnd; ++it)
                  {
                      if (!_repeatFlag && (it->second == ET_KEY_AREA_ERR))
                      {
                          repeatIndexs.push_back(_indexs[it->first]);
                      }
                      else
                      {
                          failedIndexRet[_indexs[it->first]] = (it->second == ET_KEY_AREA_ERR) ? errcode : it->second;
                      }
                  }
                  if (!repeatIndexs.empty())
                  {
                      ret = doRecall(failedIndexRet, repeatIndexs, errcode, recallmf);
                      if (ret != ET_SUCC)
                      {
                          for (size_t i = 0; i < repeatIndexs.size(); ++i)
                          {
                              failedIndexRet[repeatIndexs[i]] = errcode;
                          }
                      }
                  }
              }
              else if (ret == ET_SERVER_TYPE_ERR)
              {
                  reportException("CBError");
  
                  if (_repeatFlag || doRecall(failedIndexRet, _indexs, errcode, recallmf) != ET_SUCC)
                  {
                      for (size_t i = 0; i < _indexs.size(); ++i)
                      {
                          failedIndexRet[_indexs[i]] = errcode;
                      }
                  }
              }
              else
              {
                  for (size_t i = 0; i < _indexs.size(); ++i)
                  {
                      failedIndexRet[_indexs[i]] = errcode;
                  }
              }
  
              param->addResult(failedIndexRet);
          }
  
          if (param->_count.dec() <= 0)
          {
              ResponserPtr responser = make_responser(resmf, param->_rsp);
  
              doResponse(ET_SUCC, caller, SUCC, responser);
          }
      }
  
      /**
           * 对异常回调做统一处理
           * @param ret，接口返回值
           * @param responser，负责应答的对象
           * @return void
           */
      void procExceptionCall(const int ret, const int errcode, void (*resmf)(TarsCurrentPtr, int, const MKVBatchWriteRsp &))
      {
          const string &caller = _current->getContext()[CONTEXT_CALLER];
  
          FDLOG("CBError") << caller << "|module:" << _req.moduleName << "|object:" << _objectName << "|errno:" << ret << endl;
  
          reportException("CBError");
  
          map<int, int> rspData;
          for (size_t i = 0; i < _indexs.size(); ++i)
          {
              rspData[_indexs[i]] = errcode;
          }
  
          MKWCacheBatchCallParam<Request> *param = (MKWCacheBatchCallParam<Request> *)_param.get();
  
          param->addResult(rspData);
  
          if (_param->_count.dec() <= 0)
          {
              ResponserPtr responser = make_responser(resmf, param->_rsp);
  
              doResponse(ET_SUCC, caller, SUCC, responser);
          }
      }
  
      /**
           * 对路由失败的key重试
           */
      int doRecall(map<int, int> &failedIndexRet,
                   const vector<size_t> &repeatIndexs,
                   const int errcode,
                   void (MKWCacheProxy::*recallmf)(MKWCachePrxCallbackPtr, const Request &, const map<string, string> &))
      {
          const string &moduleName = _req.moduleName;
  
          RouterTableInfo *routeTableInfo = g_app._routerTableInfoFactory->getRouterTableInfo(moduleName);
          if (!routeTableInfo)
          {
              TLOGERROR("MKWCacheBatchCallback::doRecall do not support module:" << moduleName << endl);
  
              return ET_MODULE_NAME_INVALID;
          }
  
          int ret = routeTableInfo->update();
          if (ret != ET_SUCC)
          {
              TLOGERROR("MKWCacheBatchCallback::doRecall update route table failed, module:" << moduleName << "|ret:" << ret << endl);
          }
          else
          {
              RouterTable &routeTable = routeTableInfo->getRouterTable();
  
              map<string, Request> objectReq;
              map<string, vector<size_t>> objectDataIndexs;
              const Request &req = ((MKWCacheBatchCallParam<Request> *)(_param.get()))->_req;
              for (size_t i = 0; i < repeatIndexs.size(); ++i)
              {
                  // 从路由表中获得要访问的CacheServer的节点信息
                  ServerInfo serverInfo;
                  ret = routeTable.getMaster(req.data[repeatIndexs[i]].mainKey, serverInfo);
                  if (ret != RouterTable::RET_SUCC)
                  {
                      // 如果通过key无法获取CacheServer的节点信息
                      TLOGERROR("MKWCacheBatchCallback::doRecall can not locate key:" << req.data[repeatIndexs[i]].mainKey << "|ret:" << ret << endl);
                      failedIndexRet[repeatIndexs[i]] = errcode;
  
                      continue;
                  }
  
                  const string &objectName = serverInfo.WCacheServant;
                  typename map<string, Request>::iterator it = objectReq.find(objectName);
                  if (it == objectReq.end())
                  {
                      Request req;
                      req.moduleName = moduleName;
                      pair<typename map<string, Request>::iterator, bool> p = objectReq.insert(make_pair(objectName, req));
                      it = p.first;
                  }
                  it->second.data.push_back(req.data[repeatIndexs[i]]);
                  objectDataIndexs[objectName].push_back(repeatIndexs[i]);
              }
  
              ret = ET_SUCC;
  
              if (!objectReq.empty())
              {
                  _param->_count.add(objectReq.size());
  
                  typename map<string, Request>::const_iterator it = objectReq.begin(), itEnd = objectReq.end();
                  map<string, vector<size_t>>::const_iterator itIndex = objectDataIndexs.begin();
                  for (; it != itEnd; ++it, ++itIndex)
                  {
                      try
                      {
                          MKWCachePrxCallbackPtr cb = new ClassName(_current, _param, itIndex->second, it->second, it->first, _beginTime, true);
                          MKWCachePrx prxCache;
                          if (getProxy(it->first, prxCache) == ET_SUCC)
                          {
                              //重新访问CacheServer
                              (((MKWCacheProxy *)prxCache.get())->*recallmf)(cb, it->second, map<string, string>());
                              continue;
                          }
                      }
                      catch (exception &ex)
                      {
                          TLOGERROR("MKWCacheBatchCallback::doRecall exception:" << ex.what() << endl);
                      }
                      catch (...)
                      {
                          TLOGERROR("MKWCacheBatchCallback::doRecall catch unkown exception" << endl);
                      }
  
                      reportException("CBError");
  
                      const vector<size_t> &indexs = itIndex->second;
                      for (size_t i = 0; i < indexs.size(); ++i)
                      {
                          failedIndexRet[indexs[i]] = errcode;
                      }
  
                      _param->_count.dec();
                  }
              }
          }
  
          return ret;
      }
  
    protected:
      BatchCallParamPtr _param;
      vector<size_t> _indexs;
      Request _req;
  };
  
  struct InsertMKVBatchCallback : public ProcMKWCacheBatchCallback<InsertMKVBatchReq, InsertMKVBatchCallback>, public MKWCachePrxCallback
  {
      /**
       * 构造函数
       */
      InsertMKVBatchCallback(TarsCurrentPtr &current,
                             BatchCallParamPtr &param,
                             const vector<size_t> &indexs,
                             const InsertMKVBatchReq &req,
                             const string &objectName,
                             const int64_t beginTime,
                             const bool repeatFlag = false)
          : ProcMKWCacheBatchCallback<InsertMKVBatchReq, InsertMKVBatchCallback>(current, param, indexs, req, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~InsertMKVBatchCallback() {}
  
      virtual void callback_insertMKVBatch(int ret, const MKVBatchWriteRsp &rsp)
      {
          MKWCacheBatchCallbackDo(ret, rsp, ET_SYS_ERR, &MKWCacheProxy::async_insertMKVBatch, &Proxy::async_response_insertMKVBatch);
      }
  
      virtual void callback_insertMKVBatch_exception(int ret)
      {
          MKWCacheBatchCallbackExcDo(_req.moduleName, _objectName, ret, ET_CACHE_ERR, &Proxy::async_response_insertMKVBatch);
      }
  };
  
  struct UpdateMKVBatchCallback : public ProcMKWCacheBatchCallback<UpdateMKVBatchReq, UpdateMKVBatchCallback>, public MKWCachePrxCallback
  {
      /**
           * 构造函数
           */
      UpdateMKVBatchCallback(TarsCurrentPtr &current,
                             BatchCallParamPtr &param,
                             const vector<size_t> &indexs,
                             const UpdateMKVBatchReq &req,
                             const string &objectName,
                             const int64_t beginTime,
                             const bool repeatFlag = false)
          : ProcMKWCacheBatchCallback<UpdateMKVBatchReq, UpdateMKVBatchCallback>(current, param, indexs, req, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~UpdateMKVBatchCallback() {}
  
      virtual void callback_updateMKVBatch(int ret, const MKVBatchWriteRsp &rsp)
      {
          MKWCacheBatchCallbackDo(ret, rsp, ET_SYS_ERR, &MKWCacheProxy::async_updateMKVBatch, &Proxy::async_response_updateMKVBatch);
      }
  
      virtual void callback_updateMKVBatch_exception(int ret)
      {
          MKWCacheBatchCallbackExcDo(_req.moduleName, _objectName, ret, ET_CACHE_ERR, &Proxy::async_response_updateMKVBatch);
      }
  };
  
  struct DelMKVBatchCallback : public ProcMKWCacheBatchCallback<DelMKVBatchReq, DelMKVBatchCallback>, public MKWCachePrxCallback
  {
      /**
           * 构造函数
           */
      DelMKVBatchCallback(TarsCurrentPtr &current,
                          BatchCallParamPtr &param,
                          const vector<size_t> &indexs,
                          const DelMKVBatchReq &req,
                          const string &objectName,
                          const int64_t beginTime,
                          const bool repeatFlag = false)
          : ProcMKWCacheBatchCallback<DelMKVBatchReq, DelMKVBatchCallback>(current, param, indexs, req, objectName, beginTime, repeatFlag)
      {
      }
      virtual ~DelMKVBatchCallback() {}
  
      virtual void callback_delMKVBatch(int ret, const MKVBatchWriteRsp &rsp)
      {
          assert(ret != ET_PARTIAL_FAIL);
  
          MKWCacheBatchCallbackDo(ret, rsp, WRITE_ERROR, &MKWCacheProxy::async_delMKVBatch, &Proxy::async_response_delMKVBatch);
      }
  
      virtual void callback_delMKVBatch_exception(int ret)
      {
          MKWCacheBatchCallbackExcDo(_req.moduleName, _objectName, ret, WRITE_ERROR, &Proxy::async_response_delMKVBatch);
      }
  };
  
  #endif
  
  