SimpleXMLRPCServer标准库源码学习

class SimpleXMLRPCDispatcher:

    def __init__(self, allow_none=False, encoding=None):
        self.funcs = {} # 经过注册的 RPC 方法映射关系：{ 方法名 : 方法对象 }
        self.instance = None # 经过注册的实例（其方法都会被当作为 RPC 方法）
        self.allow_none = allow_none
        self.encoding = encoding

    def register_instance(self, instance, allow_dotted_names=False):
        """ 注册一个实例，该实例的所有方法都会被当作 RPC 方法 

        .. warning::

           使用 `allow_dotted_names` 参数可能会导致安全问题，

           调用者可以访问到实例所在模块的全局变量，甚至可以执行任意代码。

        """
        self.instance = instance
        self.allow_dotted_names = allow_dotted_names

    def register_function(self, function, name = None):
        """ 将一个函数注册为 RPC 方法 """
        if name is None:
            name = function.__name__
        self.funcs[name] = function

    def register_introspection_functions(self):
        """ 注册 RPC 内省方法 
        1. system.listMethods
        2. system.methodHelp
        3. system.methodSignature
        """
        self.funcs.update({'system.listMethods' : self.system_listMethods,
                      'system.methodSignature' : self.system_methodSignature,
                      'system.methodHelp' : self.system_methodHelp})

    def register_multicall_functions(self):
        """ 注册 RPC 多调方法 
        使用该方法，可以一次提交多个需要执行的 RPC 方法，然后服务端一次将所有结果返回 
        这样主要为了减少请求响应次数 
        """
        self.funcs.update({'system.multicall' : self.system_multicall})

    def _marshaled_dispatch(self, data, dispatch_method = None, path = None):
        """ 反序列化客户端的请求消息体，获取其中的方法名及参数，并调用注册的对应的 PRC 方法 """
        try:
            # 具体的反序列化操作是由 `xmlrpclib` 模块来完成的，其实就是解析请求消息中的 XML，
            # 获取其中的方法名及参数 
            params, method = xmlrpclib.loads(data)

            if dispatch_method is not None:
                response = dispatch_method(method, params)
            else:
                response = self._dispatch(method, params)
            response = (response,)
            response = xmlrpclib.dumps(response, methodresponse=1,
                                       allow_none=self.allow_none, encoding=self.encoding)
        except Fault, fault:
            response = xmlrpclib.dumps(fault, allow_none=self.allow_none,
                                       encoding=self.encoding)
        except:
            # report exception back to server
            exc_type, exc_value, exc_tb = sys.exc_info()
            response = xmlrpclib.dumps(
                xmlrpclib.Fault(1, "%s:%s" % (exc_type, exc_value)),
                encoding=self.encoding, allow_none=self.allow_none,
                )

        return response

    def system_listMethods(self):
        """ 列出服务端提供的所有 RPC 方法 """
        methods = self.funcs.keys()
        if self.instance is not None:
            if hasattr(self.instance, '_listMethods'):
                methods = remove_duplicates(
                        methods + self.instance._listMethods()
                    )
            elif not hasattr(self.instance, '_dispatch'):
                methods = remove_duplicates(
                        methods + list_public_methods(self.instance)
                    )
        methods.sort()
        return methods

    def system_methodSignature(self, method_name):
        """ 返回函数名对应函数的签名（参数类型列表）
        但由于 Python 函数的参数不需要指定其类型，所以函数签名功能也无法实现 
        """
        return 'signatures not supported'

    def system_methodHelp(self, method_name):
        """ 返回 RPC 方法的帮助文档 (docstring)"""
        method = None
        if method_name in self.funcs:
            method = self.funcs[method_name]
        elif self.instance is not None:
            if hasattr(self.instance, '_methodHelp'):
                return self.instance._methodHelp(method_name)
            elif not hasattr(self.instance, '_dispatch'):
                try:
                    method = resolve_dotted_attribute(
                                self.instance,
                                method_name,
                                self.allow_dotted_names
                                )
                except AttributeError:
                    pass

        if method is None:
            return ""
        else:
            import pydoc
            return pydoc.getdoc(method)

    def system_multicall(self, call_list):
        """system.multicall([{'methodName': 'add', 'params': [2, 2]}, ...]) => [[4], ...]
        """
        results = []
        for call in call_list:
            method_name = call['methodName']
            params = call['params']

            try:
                results.append([self._dispatch(method_name, params)])
            except Fault, fault:
                results.append(
                    {'faultCode' : fault.faultCode,
                     'faultString' : fault.faultString}
                    )
            except:
                exc_type, exc_value, exc_tb = sys.exc_info()
                results.append(
                    {'faultCode' : 1,
                     'faultString' : "%s:%s" % (exc_type, exc_value)}
                    )
        return results

    def _dispatch(self, method, params):
        """ 分发 RPC 方法，本质就是根据传入的方法名及参数来调用内部已注册的 RPC 方法 

        .. note::

           如果同时注册了函数和实例，那么函数的优先级比实例中的方法高 
        """
        func = None
        try:
            func = self.funcs[method]
        except KeyError:
            if self.instance is not None:
                if hasattr(self.instance, '_dispatch'):
                    return self.instance._dispatch(method, params)
                else:
                    try:
                        func = resolve_dotted_attribute(
                            self.instance,
                            method,
                            self.allow_dotted_names
                            )
                    except AttributeError:
                        pass

        if func is not None:
            return func(*params)
        else:
            raise Exception('method "%s" is not supported' % method)

class SimpleXMLRPCRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
    # RPC 路径上面在简介 XML-RPC 时就提到过 
    rpc_paths = ('/', '/RPC2')

    # 响应数据的最大传输单元，如果原始响应数据超过该值，则尝试用 GZIP 进行压缩 
    encode_threshold = 1400 #a common MTU

    # Nagle 算法会让数据发送延时，对于 RPC 这种实时应用，需要禁用它。
    # 但是一旦禁用 Nagle 算法，任何大小的数据包就会立即发送，
    # 这样就需要上层来控制每次尽量发送尽可能多的数据包 
    # 其实就是关闭底层的 Nagle 算法，然后上层自己完成 Nagle 算法的类似功能。
    wbufsize = -1
    disable_nagle_algorithm = True

    # a re to match a gzip Accept-Encoding
    aepattern = re.compile(r"""
                            \s* ([^\s;]+) \s*            #content-coding
                            (;\s* q \s*=\s* ([0-9\.]+))? #q
                            """, re.VERBOSE | re.IGNORECASE)

    def accept_encodings(self):
        r = {}
        ae = self.headers.get("Accept-Encoding", "")
        for e in ae.split(","):
            match = self.aepattern.match(e)
            if match:
                v = match.group(3)
                v = float(v) if v else 1.0
                r[match.group(1)] = v
        return r

    def is_rpc_path_valid(self):
        """ 验证 RPC 路径 """
        if self.rpc_paths:
            return self.path in self.rpc_paths
        else:
            return True

    def do_POST(self):
        """ 所有的 XML-PRC 都是 POST 请求 """
        if not self.is_rpc_path_valid():
            self.report_404()
            return

        try:
            # 这里为何要多次读取，请参见下面的 `Issue792570`
            max_chunk_size = 10*1024*1024
            size_remaining = int(self.headers["content-length"])
            L = []
            while size_remaining:
                chunk_size = min(size_remaining, max_chunk_size)
                chunk = self.rfile.read(chunk_size)
                if not chunk:
                    break
                L.append(chunk)
                size_remaining -= len(L[-1])
            data = ''.join(L)

            data = self.decode_request_content(data)
            if data is None:
                return #response has been sent

            # getattr(self, '_dispatch', None)
            # 是为了兼容以前的版本，以前的版本的分发方法 `_dispatch` 是在本类中的。
            response = self.server._marshaled_dispatch(
                    data, getattr(self, '_dispatch', None), self.path
                )
        except Exception, e: # This should only happen if the module is buggy
            # internal error, report as HTTP server error
            self.send_response(500)

            # Send information about the exception if requested
            if hasattr(self.server, '_send_traceback_header') and \
                    self.server._send_traceback_header:
                self.send_header("X-exception", str(e))
                self.send_header("X-traceback", traceback.format_exc())

            self.send_header("Content-length", "0")
            self.end_headers()
        else:
            # 如果返回的数据太大，则尝试进行 GZIP 压缩 
            self.send_response(200)
            self.send_header("Content-type", "text/xml")
            if self.encode_threshold is not None:
                if len(response) > self.encode_threshold:
                    q = self.accept_encodings().get("gzip", 0)
                    if q:
                        try:
                            response = xmlrpclib.gzip_encode(response)
                            self.send_header("Content-Encoding", "gzip")
                        except NotImplementedError:
                            pass
            self.send_header("Content-length", str(len(response)))
            self.end_headers()
            self.wfile.write(response)

    def decode_request_content(self, data):
        """ 解码 GZIP 格式的请求消息，具体参见下面的 `HTTP 压缩 ` """
        encoding = self.headers.get("content-encoding", "identity").lower()
        if encoding == "identity": # 原始数据，未经过压缩 
            return data
        if encoding == "gzip": # 经过 GZIP 格式压缩的数据 
            try:
                return xmlrpclib.gzip_decode(data)
            except NotImplementedError:
                self.send_response(501, "encoding %r not supported" % encoding)
            except ValueError:
                self.send_response(400, "error decoding gzip content")
        else:
            self.send_response(501, "encoding %r not supported" % encoding)
        self.send_header("Content-length", "0")
        self.end_headers()

    def report_404(self):
        self.send_response(404)
        response = 'No such page'
        self.send_header("Content-type", "text/plain")
        self.send_header("Content-length", str(len(response)))
        self.end_headers()
        self.wfile.write(response)

    def log_request(self, code='-', size='-'):
        if self.server.logRequests:
            BaseHTTPServer.BaseHTTPRequestHandler.log_request(self, code, size)

class SimpleXMLRPCServer(SocketServer.TCPServer,
                         SimpleXMLRPCDispatcher):

    allow_reuse_address = True

    _send_traceback_header = False

    def __init__(self, addr, requestHandler=SimpleXMLRPCRequestHandler,
                 logRequests=True, allow_none=False, encoding=None, bind_and_activate=True):
        self.logRequests = logRequests

        SimpleXMLRPCDispatcher.__init__(self, allow_none, encoding)
        SocketServer.TCPServer.__init__(self, addr, requestHandler, bind_and_activate)

        # 参见下面的 `Issue1222790`
        if fcntl is not None and hasattr(fcntl, 'FD_CLOEXEC'):
            flags = fcntl.fcntl(self.fileno(), fcntl.F_GETFD)
            flags |= fcntl.FD_CLOEXEC
            fcntl.fcntl(self.fileno(), fcntl.F_SETFD, flags)

class MultiPathXMLRPCServer(SimpleXMLRPCServer):
    def __init__(self, addr, requestHandler=SimpleXMLRPCRequestHandler,
                 logRequests=True, allow_none=False, encoding=None, bind_and_activate=True):

        SimpleXMLRPCServer.__init__(self, addr, requestHandler, logRequests, allow_none,
                                    encoding, bind_and_activate)
        self.dispatchers = {} # 该字典维护了多个 Dispatcher 实例 
        self.allow_none = allow_none
        self.encoding = encoding

    def add_dispatcher(self, path, dispatcher):
        self.dispatchers[path] = dispatcher
        return dispatcher

    def get_dispatcher(self, path):
        return self.dispatchers[path]

    def _marshaled_dispatch(self, data, dispatch_method = None, path = None):
        try:
            response = self.dispatchers[path]._marshaled_dispatch(
               data, dispatch_method, path)
        except:
            # report low level exception back to server
            # (each dispatcher should have handled their own
            # exceptions)
            exc_type, exc_value = sys.exc_info()[:2]
            response = xmlrpclib.dumps(
                xmlrpclib.Fault(1, "%s:%s" % (exc_type, exc_value)),
                encoding=self.encoding, allow_none=self.allow_none)
        return response

class CGIXMLRPCRequestHandler(SimpleXMLRPCDispatcher):
    """Simple handler for XML-RPC data passed through CGI."""

    def __init__(self, allow_none=False, encoding=None):
        SimpleXMLRPCDispatcher.__init__(self, allow_none, encoding)

    def handle_xmlrpc(self, request_text):
        response = self._marshaled_dispatch(request_text)

        print 'Content-Type: text/xml'
        print 'Content-Length: %d' % len(response)
        print
        sys.stdout.write(response)

    def handle_get(self):
        """ 在博文开关就说明了 XML-RCP 只使用 POST 方法，
        所以如果使用 GET 方法调用 CGI 脚本，就返回 400 错误 
        """
        code = 400
        message, explain = \

        BaseHTTPServer.BaseHTTPRequestHandler.responses[code]
        response = BaseHTTPServer.DEFAULT_ERROR_MESSAGE % \
            {
             'code' : code,
             'message' : message,
             'explain' : explain
            }
        print 'Status: %d %s' % (code, message)
        print 'Content-Type: %s' % BaseHTTPServer.DEFAULT_ERROR_CONTENT_TYPE
        print 'Content-Length: %d' % len(response)
        print
        sys.stdout.write(response)

    def handle_request(self, request_text = None):
        """ 入口方法，

        * 默认从标准输入读入请求消息中的 XML 格式数据 

        * 也支持从函数的参数传入请求消息中的 XML 格式数据 

        然后对该数据进行解析，取出其中的方法名和参数，并进行 RPC 方法的分发 
        """
        if request_text is None and \
            os.environ.get('REQUEST_METHOD', None) == 'GET':
            self.handle_get()
        else:
            # POST data is normally available through stdin
            try:
                length = int(os.environ.get('CONTENT_LENGTH', None))
            except (TypeError, ValueError):
                length = -1
            if request_text is None:
                request_text = sys.stdin.read(length)

            self.handle_xmlrpc(request_text)