httpserver_test.py 文件源码

def get_app(self):
        class BufferedHandler(RequestHandler):
            def put(self):
                self.write(str(len(self.request.body)))

        @stream_request_body
        class StreamingHandler(RequestHandler):
            def initialize(self):
                self.bytes_read = 0

            def prepare(self):
                if 'expected_size' in self.request.arguments:
                    self.request.connection.set_max_body_size(
                        int(self.get_argument('expected_size')))
                if 'body_timeout' in self.request.arguments:
                    self.request.connection.set_body_timeout(
                        float(self.get_argument('body_timeout')))

            def data_received(self, data):
                self.bytes_read += len(data)

            def put(self):
                self.write(str(self.bytes_read))

        return Application([('/buffered', BufferedHandler),
                            ('/streaming', StreamingHandler)])