def handle_cmd(self, frame):
        """
        Method:    handle_cmd
        Description: 处理订阅者加入分组的命令消息
        Parameter: 
            frame: AppFrame
        Return: 
        Others: 
        """
        tracelog.info(self)
                
        result = basic_rep_to_web.BasicRepToWeb()
        result.init_all_attr()
        tracelog.info('subscriber cfg worker recv group frame: %s' % frame)
        
        self._req_frame = frame
        buf = frame.get_data()
        
        try:
            assoc_req = msg_params_def.PortalSubscriberGroupAssociateReq.deserialize(buf)
        except:
            result.return_code = err_code_mgr.ERR_PORTAL_DESERIALIZE_ERROR
            result.description = err_code_mgr.get_error_msg(err_code_mgr.ERR_PORTAL_DESERIALIZE_ERROR,
                                                            cmd = 'PORTAL_SUBSCRIBER_GROUP_ASSOCIATE',
                                                            param_name = 'PortalSubscriberGroupAssociateReq')
            result.user_session = ''
            self.get_worker().get_app().send_ack_dispatch(frame, (result.serialize(), ))
            return

        self._message = assoc_req
        # 检查订阅者是否存在
        subs = self.get_worker().get_app().get_mit_manager().rdm_find('Subscriber',
                                                                      subscriber_open_id = assoc_req.subscriber_open_id)   

        if len(subs) == 0:
            result.return_code = err_code_mgr.ERR_PORTAL_SUBSCRIBER_NOT_EXISTS
            result.description = err_code_mgr.get_error_msg(err_code_mgr.ERR_PORTAL_SUBSCRIBER_NOT_EXISTS)

            result.prepare_for_ack(assoc_req, result.return_code, result.description)
            self.get_worker().get_app().send_ack_dispatch(frame, (result.serialize(), ))
            return
        else:
            # 检查订阅者关联的分组是否都存在
            group_ids = assoc_req.group_ids
            for gid in group_ids:
                grps = self.get_worker().get_app().get_mit_manager().rdm_find('Group',
                                                                              group_id = int(gid))
                if len(grps) == 0:
                    result.return_code = err_code_mgr.ERR_PORTAL_GROUP_NOT_EXISTS
                    result.description = err_code_mgr.get_error_msg(err_code_mgr.ERR_PORTAL_GROUP_NOT_EXISTS)
        
                    result.prepare_for_ack(assoc_req, result.return_code, result.description)
                    self.get_worker().get_app().send_ack_dispatch(frame, (result.serialize(), ))
                    return
            
            sub_frame = bf.AppFrame()
            sub_frame.set_cmd_code(cmd_code_def.CLOUD_PORTAL_SUB_GROUP_ASSOC_MSG)
            sub_frame.set_receiver_pid(self.get_worker().get_pid("SubscriberManApp"))
            sub_frame.add_data(buf)
            self.wait_for_ack(sub_frame, 5)

    def rmv_node(self, ip):
        """
        Method: rmv_node
        Description: 删除指定的节点
        Parameter: 
            ip: 指定的节点的ip
        Return: 错误码，错误信息
        Others: 
        """

        online_err = (err_code_mgr.ER_CANNOT_RMV_ONLINE_CLUSTER_NODE
                        , err_code_mgr.get_error_msg(err_code_mgr.ER_CANNOT_RMV_ONLINE_CLUSTER_NODE))
                        
        with self.__lock:
            # 只允许删除离线的节点
            if ip == self.__cluster_cfg_info.my_inner_ip:
                return online_err

            for node in self.__other_nodes:
                if node.get_ip() == ip and node.is_online():
                    return online_err
            
            # 先删除mit中的信息
            ret_code, err_msg = self.__mit.rmv_node(ip)

            if ret_code == 0:
                # 删除内存中的信息
                self.__rmv_node(ip)
                
                tracelog.info("remvoe node %s." % ip)
            else:
                tracelog.error("remvoe node %s failed." % ip)

        return ret_code, err_msg

    def enter_state(self):
        """
        Method: enter_state
        Description: 
        Parameter: 
        Return: 
        Others: 
        """

        tracelog.info ('member %s enter %s state' % (self._processor.get_target().get_id(), fsm_def.MEMBER_DELIVERY_STATE))
        if self._processor.get_target().get_old_state() == fsm_def.MEMBER_DELIVERY_STATE:
            return

        # 如果会员与微信订阅者已绑定，通知订阅者自选菜单
        if self._processor.get_target().get_spec().subscriber_open_id is not None and len(self._processor.get_target().get_spec().subscriber_open_id) > 0:
            # 构造推送消息发给SubscriberManApp
            push_msg = msg_params_def.CloudPortalTextPushMessage()
            push_msg.init_all_attr()
            push_msg.subscriber_open_ids = [self._processor.get_target().get_spec().subscriber_open_id]
            raw_txt = msg_params_def.PORTAL_TXT_DELIVERY_EXPIRED_NOTIFY_TXT.decode('gbk').encode('utf-8') % (self._processor.get_target().get_spec().name, 
                                                                                                     self._processor.get_target().get_id())
            push_msg.text_msg = raw_txt
            push_frame = bf.AppFrame()
            push_frame.set_cmd_code(cmd_code_def.CLOUD_PORTAL_TEXT_PUSH_MSG)
            push_frame.add_data(push_msg.serialize())
            self._processor.get_worker().dispatch_frame_to_process_by_pid(self._processor.get_worker().get_pid("SubscriberManApp"), push_frame)

def reinforce_telnet():
    """
    Function: reinforce_telnet
    Description: 关闭telnet服务
    Parameter: 无
    Return: return_code，0代表成功，非0代表失败
    Others: 无
    """
    
    return_code = 0
    for u in xrange(3):
        if os.system("service telnet stop") != 0:
            tracelog.error('Fail to stop service telnet and have tried %u' %(u))
            time.sleep(1)
        else:
            tracelog.info('telnet service stop')
            break
    else:
        return_code = 1
    for u in xrange(3):
        if os.system("chkconfig telnet off") != 0:
            tracelog.error('Fail to set service telnet off and have tried %u' %(u))
            time.sleep(1)
        else:
            tracelog.info('telnet service set off')
            break  
    else:
        return_code = 1 
            
    return return_code

    def _ready_for_work(self):
        """
        Method:    _ready_for_work
        Description: 线程工作前的初始化函数
        Parameter: 无
        Return: 
            0: 成功
            非0: 失败
        Others: 
        """

        self.__doing_what = WorkThread.doing_ready_for_work
        
        for wrkr in self.__workers:
            try:
                ret = wrkr.ready_for_work()
            except:
                tracelog.exception("worker %s ready_for_work failed." % wrkr.get_name())
                return -1
                
                
            if ret == 0:
                tracelog.info("%s is ready for work." % wrkr.get_name())
            else:
                tracelog.error("%s is not ready." % wrkr.get_name())
                return ret
                
        return 0

    def __when_now_slave(self):
        """
        Method: __when_now_slave
        Description: 当处于slave的处理函数
        Parameter: 无
        Return: 
        Others: 
        """

        # 没有收到master的查询，并且没有其他节点，或没有收到ip更小的应答
        # 那么转为master   
        with self.__lock:
            for node in self.__other_nodes:
                if not node.is_enable():
                    continue
                
                if node.is_role_master():
                    node.check_heartbeat()
                    
                    if node.is_online():   
                        old_master_ip = self.__mater_node_info.get_ip()
                        if self.__mater_node_info.update(node.get_ip(), node.get_start_time()):
                            self.__cluster_node.on_master_change(old_master_ip, self.__mater_node_info.get_ip())
                            
                        return
                

        tracelog.info("the master cluster node is offline.")

        
        # 切换为无主状态
        self.__change_state(CLUSTER_STATE_NO_MASTER)
        self.__reset_query_counter(True)
        self.reload_nodes()

    def enter_state(self):
        """
        Method: enter_state
        Description: 
        Parameter: 
        Return: 
        Others: 
        """

        tracelog.info ('subscriber %s enter %s state' % (self._processor.get_target().get_id(), fsm_def.SUBSCRIBER_INIT_STATE))
        if self._processor.get_target().get_old_state() == fsm_def.SUBSCRIBER_INIT_STATE:
            return

        last_frame = self._processor.get_target().get_last_frame()
        wx_sub_msg = msg_params_def.WXPushEventMessage.deserialize(last_frame.get_data())
        
        content = self._processor.get_worker().get_helptips()
        content += "\r\n" + msg_params_def.WX_TXT_SUBSCRIBE_TIPS.decode('gbk').encode('utf-8') % self._processor.get_target().get_spec().subscribe_seq_no
        
        reply_msg_type = msg_params_def.WX_MSG_TYPE_TEXT
        
        sub_welcome_msg = msg_params_def.WXReplyTextMessage()
        sub_welcome_msg.init_all_attr()
        sub_welcome_msg.subscriber_open_id = wx_sub_msg.subscriber_open_id
        sub_welcome_msg.public_account_id = wx_sub_msg.public_account_id
        sub_welcome_msg.create_time = str(int(time.time()))
        sub_welcome_msg.msg_type = reply_msg_type
        sub_welcome_msg.content = content
        sub_welcome_msg.func_flag = '0'
        
        self._processor.get_worker().get_app().send_ack_dispatch(last_frame, (reply_msg_type, sub_welcome_msg.serialize()))
        self._processor.get_target().free_frame()

    def start_listen(self):
        """
        Method: start_listen
        Description: 开始监听
        Parameter: 无
        Return:错误码
        Others: 
        """

        if self._ssl_option is None:
            ret = self.bind("", self._port)
        else:
            ret = self.bind_with_ssl(""
                                , self._port
                                , self._ssl_option.get("key_file", "")
                                , self._ssl_option["cert_file"]
                                , self._ssl_option["proto_version"])

        if ret == 0:
            tracelog.info("SimpleCallAcpSrv start listen on port %d, ssl_option:%s"%(
                            self._port
                            , self._ssl_option))
        else:
            tracelog.error("SimpleCallAcpSrv listen on port %d failed, ssl_option:%s"%(
                            self._port
                            , self._ssl_option))

        return ret

    def _insert_task_to_map(self, task_type, abs_time, times, object_id, count):
        """
        Method: _insert_task_to_map
        Description: 将任务插入定时器有序队列
        Parameter: 
            abs_time: 定时时间 YYYY-MM-DD HH:MM:SS
            times: 推送重复次数
            article_id: 推送主题内容ID
        Return: 
        Others: 
        """

        #time_stru = time.strptime(abs_time, '%Y-%m-%d %H:%M:%S')
        #abs_time_f = time.mktime(time_stru)   
              
        index = 0
        items = self._time_task_map.items()
        for item in items:
            if abs_time < item[0]:
                break
            index += 1
        items.insert(index, (abs_time, (task_type, object_id, times, count)))
        self._time_task_map = collections.OrderedDict(items)

        tracelog.info('add new timer task(type %s, time %s, times %s, id %d, count %d)' % (task_type, 
                                                                                           time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(abs_time)), 
                                                                                           times, 
                                                                                           object_id, 
                                                                                           count))

    def _login(self):
        self._requests_session.cookies.clear_session_cookies()

        self._requests_session.get(WINDELN_URL, verify = True)
        self._requests_session.get(WINDELN_LOGIN_REFERER, verify = True)

        header = {}
        
        header['Referer'] = WINDELN_LOGIN_REFERER
        header['Accept'] = 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8'
        header['Accept-Encoding'] = 'gzip,deflate,sdch'
        header['Accept-Language'] = 'zh-CN,zh;q=0.8'
        header['Cache-Control'] = 'max-age=0'
        header['Connection'] = 'keep-alive'
        header['Content-Type'] = 'application/x-www-form-urlencoded'
        header['User-Agent'] = 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/27.0.1453.110 Safari/537.36'
        header['Host'] = 'www.windeln.de'
        header['Origin'] = 'https://www.windeln.de'

        p = {}
        p['login[username]'] = self._username
        p['login[password]'] = self._pwd
        body = urllib.urlencode(p)
        
        resp = self._requests_session.post(WINDELN_LOGIN_URL, verify = True, data = body, headers = header)
        print resp.url
        
        if resp.status_code == 200:
            if resp.url.find('login') < 0:
                tracelog.info('Windeln(%s) login succ!' % self._username)
                self._login_succ = True
                return True
        
        return False        

    def handle_cmd(self, frame):

        tracelog.info("receive restart ACP system command.")
        
        worker = self.get_worker()
        app = worker.get_app()
        app.soft_restart()

    def __init_sync_table_data(self):
        """
        Method: __init_sync_table_data
        Description: 初始化数据同步表，将MOC数据，从user_acp用户
                     同步到user_sync用户的表中
        Parameter: 无
        Return: 错误码
        Others: 
        """

        ret = self.__delete_old_sync_table_data()
        if ret != 0:            
            return err_code_mgr.ER_SYNC_EXPORT_DATA_FAILED

        con_pool = self.get_worker().get_app().get_conn_pool()
        with con_pool.get_connection(db_cfg_info.ORACLE_DEFAULT_CON_NAME, False) as db_con:   
            db_con.begin()
            ret = self.__lock_table(db_con)
            if ret != 0:
                db_con.rollback()
                return err_code_mgr.ER_SYNC_EXPORT_DATA_FAILED
                
            tracelog.info("lock table ok.")
                         
            ret = self.__update_sync_table()
            if ret != 0:
                db_con.rollback()
                return err_code_mgr.ER_SYNC_EXPORT_DATA_FAILED

            db_con.commit()
            
        return ret

    def handle_cmd(self, frame):
        """
        Method: handle_cmd
        Description: 注销请求处理函数，调用name_mit_manager的相应函数进行处理，然后更新本地缓存，再广播到其他APP。
        Parameter: 
            frame: 注销请求消息，数据区为AppUnRegisterRequest
        Return: 
        Others: 
        """

        buf = frame.get_data()
        result = bf.AppUnRegisterResponse()
        result.init_all_attr()
        app = self.get_worker().get_app()
        try:
            un_reg_req = bf.AppUnRegisterRequest.deserialize(buf)
            tracelog.info("unregist name service:%r" % un_reg_req)
        except Exception:            
            result.return_code = err_code_mgr.ER_INVALID_DESERIALIZE_STRING_ERROR
            result.description = err_code_mgr.get_error_msg(err_code_mgr.ER_INVALID_DESERIALIZE_STRING_ERROR
                                                            , cmd='UnRegisterApp'
                                                            , param_name='AppUnRegisterRequest')            
            app.send_ack(frame, result.serialize())
            tracelog.error('%s\n%s'%(result.description, buf))
            return
            
        result = self.get_worker().unregister_app(un_reg_req)        

        if un_reg_req.need_reponse is True:
            app.send_ack(frame, result.serialize())
        
        return

    def _on_round_over(self, round_id, r):        
        """
        Method: _on_round_over
        Description: 接收业务模块的响应，构造XML返回给微信
        Parameter: 
            round_id: 
            r: 
        Return: 
        Others: 
        """
        frame = r.get_response_frame()
        msg_type = frame.get_data()
        msg_buf = frame.get_data(1)

        wx_rpl_msg_proc_dict = {msg_params_def.WX_MSG_TYPE_TEXT: WXRplTextMsgProc,
                                msg_params_def.WX_MSG_TYPE_MUSIC: WXRplMusicMsgProc,
                                msg_params_def.WX_MSG_TYPE_NEWS: WXRplNewsMsgProc,
                                }        
        wx_rpl_msg_proc = wx_rpl_msg_proc_dict[msg_type]
        xml_msg = wx_rpl_msg_proc(msg_buf).gen_xml()
        
        reply = msg_params_def.CommonContentRsp()
        reply.init_all_attr()
        reply.return_code = err_code_mgr.ER_SUCCESS
        reply.description = err_code_mgr.get_error_msg(err_code_mgr.ER_SUCCESS)
        reply.prepare_for_ack(self._message, reply.return_code, reply.description)          
        reply.content = xml_msg

        tracelog.info('reply msg to WX: %s' % reply.content)
        
        self.get_worker().get_app().send_ack_dispatch(self._req_frame, (reply.serialize(), ))

    def run_case(self, app, case):
        """
        Method:    run_case
        Description: 运行一个用例
        Parameter: 
            case: 一个用例, instance of ALECase
        Return: 
            0: 成功
            非0: 失败
        Others: 
        """

        tracelog.info("run case %s..." % str(case))

        ret = self.find_case_entity(case)    
        if ret != 0:
            return -1

        case.case_entity

        frame = bf.AppFrame()
        frame.set_cmd_code(case.case_entity.get_cmd_code())
        frame.add_data(case.case_entity.gen_request())
        frame.set_receiver_pid(app.get_pid("AleGate"))

        ack_frames = bf.rpc_request(frame, case.case_entity.get_timeout())
        if len(ack_frames) == 0:
            tracelog.error("wait for response timeout. case:%s" % str(case)) 
        else:
            case.case_entity.handle_ack(ack_frames[0].get_data())
        
        return 0

    def on_msg_received(self, url_or_srv_name, msg):
        #print "on_msg_received:", msg.get_cmd_code(), url_or_srv_name, msg.get_data(), msg.get_msg_id()

        cmd_code = msg.get_cmd_code()
        
        if cmd_code == bf.CMD_QUERY_CLUSTER_MASTER_IP:            
            rep = bf.QueryClusterMasterIpResponse()
            rep.ip = self.__name_server.get_cluster_master_ip()
        elif cmd_code == bf.REGISTER_NAME_COMMAND:
            try:
                req = bf.AppRegisterRequest.deserialize(msg.get_data())
            except:
                tracelog.exception("AppRegisterRequest deserialize failed.")
                return

            rep = self.__name_server.register_app(req)

            if rep.return_code == 0:
                tracelog.info("app regist name service: %r, pid:%r" % (req, rep.app_info.pid))
            else:
                tracelog.error("app regist name service failed: %r" % req)

        ack_msg = pycallacp.AcpMessage(pycallacp.CMD_ACK_MSG, rep.serialize())
        ack_msg.set_msg_id(msg.get_msg_id())
        self._callacp_inst.send(url_or_srv_name, ack_msg)

    def __switch_to_master(self, state):
        """
        Method: __switch_to_master
        Description: 将角色切换到master
        Parameter: 
            state: 状态
        Return: 
        Others: 
        """

        # 切换到master

        old_role = self.__role
        old_state = self.__state
        self.__mater_node_info.update(self.__cluster_cfg_info.my_inner_ip, self.__start_time)
        
        self.__role = CLUSTER_ROLE_MASTER
        self.__state = state
        tracelog.info("the current cluster node %s switch to master. state:%d" % (
                          self.__cluster_cfg_info.my_inner_ip
                        , state))

        ret_code  = self.__bind_virtual_ip(True)
        if ret_code != 0:
            tracelog.error("bind virtual ip faild. ret_code:%d" % ret_code)

        self.__cluster_node.on_state_change(old_role, old_state, self.__role, state)

        # 进入master状态后，重新设置其他节点的状态
        self.__reset_query_counter(True)

    def __switch_to_slave(self):
        """
        Method: __switch_to_slave
        Description: 将角色切换到slave
        Parameter: 无
        Return: 
        Others: 
        """

        # 切换到slave

        old_role = self.__role
        old_state = self.__state
        
        self.__role = CLUSTER_ROLE_SLAVE
        self.__state = CLUSTER_STATE_NORMAL
        tracelog.info("the current cluster node %s switch to slave. state:%d" % (
                            self.__cluster_cfg_info.my_inner_ip
                           , state))

        ret_code  = self.__unbind_virtual_ip(True)
        if ret_code != 0:
            tracelog.error("unbind virtual ip faild. ret_code:%d" % ret_code)
            
        self.__cluster_node.on_state_change(old_role, old_state, self.__role, state)

    def handle_cmd(self, frame):
        """
        Method:    handle_cmd
        Description: 处理订阅者管理模块初始化消息
        Parameter: 
            frame: AppFrame
        Return: 
        Others: 
        """
        tracelog.info(self)
        
        subs = self.get_worker().get_app().get_mit_manager().rdm_find("Subscriber")
        for sub_moc in subs:
            grps = self.get_worker().get_app().get_mit_manager().lookup_attrs("Subscriber", ['group_ids'], subscriber_open_id = sub_moc.subscriber_open_id)
            sub = subscriber_def.Subscriber(sub_moc, grps[0][0], fsm_def.SUBSCRIBER_SESSION_STATE)

            if sub_moc.assoc_member_id is not None and len(sub_moc.assoc_member_id) > 0:
                mbrs = self.get_worker().get_app().get_mit_manager().rdm_find("Member", member_id = sub_moc.assoc_member_id)
                if len(mbrs) > 0:
                    sub.set_assoc_member(mbrs[0])
            
            processor = fsm_def.StateProcessor(sub, self.get_worker())
            processor.register_state_handler(fsm_def.SUBSCRIBER_INIT_STATE, subscriber_state_man.SubInitStateHandler(processor))
            processor.register_state_handler(fsm_def.SUBSCRIBER_SESSION_STATE, subscriber_state_man.SubSessionStateHandler(processor))
            
            self.get_worker().get_state_manager().add_processor(processor)          

 def get_pid(self, service_name, strategy=local_const_def.FIRSTLOCAL_PID):
     """
     Method:    get_pid
     Description: 根据选择策略获取指定service_name的进程的pid
     Parameter: 
         service_name: 服务名
     Return: 
         -1: 指定的进程不存在
         其他: 进程的pid
     Others: 
                     选择策略有多种：
     FIRSTLOCAL_PID，优选本机的进程，本机的进程不存在，则随机选择其他系统的进程
     ONLYLOCAL_PID,只选择本机的进程，本机的进程不存在，返回查找不到的信息
     RANDOM_PID,随机选择
     MASTER_PID,选择EAU或者IMC MASTER上的进程
     """
     with self.__mutex:
         pid = self.__app_pid.get_pid(service_name, strategy)
         if pid==0:
             #tracelog.info('get pid failed. service_name: %s all_name_pids: %s'%(
             #                service_name
             #                , self.__app_pid.get_all_name_pids()
             #                ))
             tracelog.info('get pid failed. service_name: %s'%(service_name))
     return pid