我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用tornado.httpserver.HTTPServer()。
def test_method_dispatcher(): """ This function can be used to test that the MethodDispatcher is working properly. It is called automatically when this script is executed directly. """ import logging from tornado.ioloop import IOLoop from tornado.httpserver import HTTPServer from tornado.options import define, options, parse_command_line define("port", default=8888, help="Run on the given port", type=int) parse_command_line() logging.info( "Test Server Listening on http://0.0.0.0:%s/" % options.port ) http_server = HTTPServer(TestApplication()) http_server.listen(options.port) IOLoop.instance().start()
def test_missing_key(self): """A missing SSL key should cause an immediate exception.""" application = Application() module_dir = os.path.dirname(__file__) existing_certificate = os.path.join(module_dir, 'test.crt') existing_key = os.path.join(module_dir, 'test.key') self.assertRaises((ValueError, IOError), HTTPServer, application, ssl_options={ "certfile": "/__mising__.crt", }) self.assertRaises((ValueError, IOError), HTTPServer, application, ssl_options={ "certfile": existing_certificate, "keyfile": "/__missing__.key" }) # This actually works because both files exist HTTPServer(application, ssl_options={ "certfile": existing_certificate, "keyfile": existing_key, })
def test_chunked_request_body(self): # Chunked requests are not widely supported and we don't have a way # to generate them in AsyncHTTPClient, but HTTPServer will read them. self.stream.write(b"""\ POST /echo HTTP/1.1 Transfer-Encoding: chunked Content-Type: application/x-www-form-urlencoded 4 foo= 3 bar 0 """.replace(b"\n", b"\r\n")) read_stream_body(self.stream, self.stop) headers, response = self.wait() self.assertEqual(json_decode(response), {u('foo'): [u('bar')]})
def setUp(self): if IOLoop.configured_class().__name__ in ('TwistedIOLoop', 'AsyncIOMainLoop'): # TwistedIOLoop only supports the global reactor, so we can't have # separate IOLoops for client and server threads. # AsyncIOMainLoop doesn't work with the default policy # (although it could with some tweaks to this test and a # policy that created loops for non-main threads). raise unittest.SkipTest( 'Sync HTTPClient not compatible with TwistedIOLoop or ' 'AsyncIOMainLoop') self.server_ioloop = IOLoop() sock, self.port = bind_unused_port() app = Application([('/', HelloWorldHandler)]) self.server = HTTPServer(app, io_loop=self.server_ioloop) self.server.add_socket(sock) self.server_thread = threading.Thread(target=self.server_ioloop.start) self.server_thread.start() self.http_client = HTTPClient()
def start(config): SERVER_PORT = config["base_port"] SERVER_BIND_ADDRESS = config["bind_address"] task_id = process.fork_processes(4) config = { "config": config } application = web.Application([ (r"/", InfoHandler, dict(config=config)), (r"/([^/]+)", IndexHandler, dict(config=config)), (r"/([^/]+)/([^/]+)", IndexQueryHandler, dict(config=config)), (r"/([^/]+)/([^/]+)/([^/]+)", IndexQueryHandler, dict(config=config)) ]) http_server = httpserver.HTTPServer(application) http_server.add_sockets(netutil.bind_sockets(SERVER_PORT + task_id, address=SERVER_BIND_ADDRESS)) log.info("Frontend listening on %d", SERVER_PORT + task_id) IOLoop.current().start()
def start_server(app: web.Application = None, port: int = None, address: str = None, **kwargs: Any) -> HTTPServer: """Start server with ``app`` on ``localhost:port``. If port is not specified, use command line option of ``--port``. """ app = app or get_app() port = port if port is not None else config.port address = address if address is not None else config.address server = app.listen(port, address=address) app.server = server server_config['address'] = address for sock in server._sockets.values(): if sock.family == socket.AF_INET: server_config['port'] = sock.getsockname()[1] break return server
def cmd_web(unix_socket, debug): app = make_app(debug=debug) server = HTTPServer(app) if os.path.exists(unix_socket): try: r = requests.get('http+unix://{0}'.format(unix_socket.replace('/', '%2F'))) if r.text.strip() == 'atx.taskqueue': print 'Already listening' return except: print 'Unlink unix socket' os.unlink(unix_socket) socket = bind_unix_socket(unix_socket) server.add_socket(socket) IOLoop.current().start()
def __init__(self, io_loop, fail_these, expected_basename): assert io_loop is not None self.fail_these = fail_these self.expected_basename = expected_basename self._application = FakeAnacondaApplication(server=self, io_loop=io_loop) self._http = HTTPServer(self._application, io_loop=io_loop) # these would throw OSError on failure sockets = bind_sockets(port=None, address='127.0.0.1') self._port = None for s in sockets: # we have to find the ipv4 one if s.family == socket.AF_INET: self._port = s.getsockname()[1] assert self._port is not None self._http.add_sockets(sockets) self._http.start(1)
def __init__(self, project, prepare_stage, event_handler, io_loop): assert event_handler is not None assert io_loop is not None self._application = UIApplication(project, prepare_stage, event_handler, io_loop) self._http = HTTPServer(self._application, io_loop=io_loop) # these would throw OSError on failure sockets = bind_sockets(port=None, address='127.0.0.1') self._port = None for s in sockets: # we have to find the ipv4 one if s.family == socket.AF_INET: self._port = s.getsockname()[1] assert self._port is not None self._http.add_sockets(sockets) self._http.start(1)
def web(port = 23456, via_cli = False, ): """ Bind Tornado server to specified port. """ print ('BINDING',port) try: tornado.options.parse_command_line() http_server = HTTPServer(Application(), xheaders=True, ) http_server.bind(port) http_server.start(16) # Forks multiple sub-processes tornado.ioloop.IOLoop.instance().set_blocking_log_threshold(0.5) IOLoop.instance().start() except KeyboardInterrupt: print 'Exit' print ('WEB_STARTED')
def run(self): import tornado.autoreload tornado.autoreload.watch('index.wsgi') import re from tornado.web import URLSpec, StaticFileHandler # The user should not use `tornado.web.Application.add_handlers` # since here in SAE one application only has a single host, so here # we can just use the first host_handers. handlers = self.application.handlers[0][1] for prefix, path in self.static_files.iteritems(): pattern = re.escape(prefix) + r"(.*)" handlers.insert(0, URLSpec(pattern, StaticFileHandler, {"path": path})) os.environ['sae.run_main'] = '1' import tornado.ioloop from tornado.httpserver import HTTPServer server = HTTPServer(self.application, xheaders=True) server.listen(self.conf.port, self.conf.host) tornado.ioloop.IOLoop.instance().start()
def test_unix_socket(self): sockfile = os.path.join(self.tmpdir, "test.sock") sock = netutil.bind_unix_socket(sockfile) app = Application([("/hello", HelloWorldRequestHandler)]) server = HTTPServer(app, io_loop=self.io_loop) server.add_socket(sock) stream = IOStream(socket.socket(socket.AF_UNIX), io_loop=self.io_loop) stream.connect(sockfile, self.stop) self.wait() stream.write(b("GET /hello HTTP/1.0\r\n\r\n")) stream.read_until(b("\r\n"), self.stop) response = self.wait() self.assertEqual(response, b("HTTP/1.0 200 OK\r\n")) stream.read_until(b("\r\n\r\n"), self.stop) headers = HTTPHeaders.parse(self.wait().decode('latin1')) stream.read_bytes(int(headers["Content-Length"]), self.stop) body = self.wait() self.assertEqual(body, b("Hello world"))
def main(): global config, output_wechat, app config = ConfigParser() config.read('config') from .utility import output_wechat app = Application() parse_command_line() try: from .views import app http_server = HTTPServer(app) http_server.listen(config['base']['port']) IOLoop.current().start() except Exception as e: print(e) finally: with open('config', 'w') as f: config.write(f)
def test_missing_key(self): """A missing SSL key should cause an immediate exception.""" application = Application() module_dir = os.path.dirname(__file__) existing_certificate = os.path.join(module_dir, 'test.crt') self.assertRaises(ValueError, HTTPServer, application, ssl_options={ "certfile": "/__mising__.crt", }) self.assertRaises(ValueError, HTTPServer, application, ssl_options={ "certfile": existing_certificate, "keyfile": "/__missing__.key" }) # This actually works because both files exist HTTPServer(application, ssl_options={ "certfile": existing_certificate, "keyfile": existing_certificate })
def test_unix_socket(self): sockfile = os.path.join(self.tmpdir, "test.sock") sock = netutil.bind_unix_socket(sockfile) app = Application([("/hello", HelloWorldRequestHandler)]) server = HTTPServer(app, io_loop=self.io_loop) server.add_socket(sock) stream = IOStream(socket.socket(socket.AF_UNIX), io_loop=self.io_loop) stream.connect(sockfile, self.stop) self.wait() stream.write(b"GET /hello HTTP/1.0\r\n\r\n") stream.read_until(b"\r\n", self.stop) response = self.wait() self.assertEqual(response, b"HTTP/1.0 200 OK\r\n") stream.read_until(b"\r\n\r\n", self.stop) headers = HTTPHeaders.parse(self.wait().decode('latin1')) stream.read_bytes(int(headers["Content-Length"]), self.stop) body = self.wait() self.assertEqual(body, b"Hello world") stream.close() server.stop()
def main(app: Flask, tornado: bool=False) -> None: """Run the Flask application.""" with app.app_context(): db.create_all() print('[OK] Database creation complete.') if tornado: from tornado.wsgi import WSGIContainer from tornado.httpserver import HTTPServer from tornado.ioloop import IOLoop http_server = HTTPServer(WSGIContainer(app)) http_server.listen(int(app.config['INIT_PORT'])) IOLoop.instance().start() else: socketio.run(app, **app.config['INIT'])
def test_chunked_request_body(self): # Chunked requests are not widely supported and we don't have a way # to generate them in AsyncHTTPClient, but HTTPServer will read them. self.stream.write(b"""\ POST /echo HTTP/1.1 Transfer-Encoding: chunked Content-Type: application/x-www-form-urlencoded 4 foo= 3 bar 0 """.replace(b"\n", b"\r\n")) read_stream_body(self.stream, self.stop) headers, response = self.wait() self.assertEqual(json_decode(response), {u'foo': [u'bar']})
def start(self): def go(app, port, data={}): from httpretty import HTTPretty HTTPretty.disable() http = HTTPServer(app) HTTPretty.disable() http.listen(int(port)) IOLoop.instance().start() app = self.get_handlers() data = {} args = (app, self.port, data) HTTPretty.disable() self.process = Process(target=go, args=args) self.process.start() time.sleep(1)
def _patch_httpserver(self): """ ??httpserver?xheader????gunicorn????xheaders?? :return: """ httpserver = sys.modules["tornado.httpserver"] try: xhs = settings.XHEADERS except: xhs = True class tornadopyHTTPServer(httpserver.HTTPServer): def __init__(self, request_callback, xheaders=xhs, **kwargs): super(tornadopyHTTPServer, self).__init__(request_callback, xheaders=xheaders, **kwargs) httpserver.HTTPServer = tornadopyHTTPServer sys.modules["tornado.httpserver"] = httpserver
def load_httpserver(self, sockets=None, **kwargs): if not sockets: from tornado.netutil import bind_sockets if settings.IPV4_ONLY: import socket sockets = bind_sockets(options.port, options.address, family=socket.AF_INET) else: sockets = bind_sockets(options.port, options.address) http_server = tornado.httpserver.HTTPServer(self.application, **kwargs) http_server.add_sockets(sockets) self.httpserver = http_server return self.httpserver
