def subscribeConfig()

in leda_python/mbus.py [0:0]

• 20 lines of code
• 3 McCabe index (conditional complexity)

    def subscribeConfig(self, key, type, reply_cb=None, error_cb=None):
        ''' 订阅配置

		:param key[string]: 配置名
		:param type[int]: 订阅类型（0.拥有者，1.观察者）
		:param reply_cb: async reply call back
		:param error_cb: async error call back
		:return:
		'''

        driverWKN = self.getName()

        if (None == self.mbusNameObj):
            raise ledaException.LedaException(" mbus name object is None")

        try:
            interfaceHandle = self.getRemoteInterface(mbusConfig.DMP_CM_WKN,
                                                      mbusConfig.DMP_CM_OBJ_PATH, mbusConfig.DMP_CM_INTERFACE)

            interfaceHandle.subscribe_config(driverWKN, key, type, reply_handler=reply_cb, error_handler=error_cb,
                                             timeout=mbusConfig.METHOD_ACK_TIMEOUT)

        except dbus.exceptions.DBusException as err:
            _logger.warning('%s', err)
            raise ledaException.LedaRPCMethodException("rpc method: subscribe_config failed",
                                                       ledaException.LEDA_ERROR_FAILED)