def startWebService():
try:
server = HTTPServer( ('localhost',8080), MyHandler)
print("started http server....")
server.serve_forever()
except KeyboardInterrupt:
print ("shutdown web server")
server.socket.close() # server ?????.
python类HTTPServer()的实例源码
def __init__(self, host, port, handler, context, docroot, index, auth=None, realm=None):
try:
BaseHTTPServer.HTTPServer.__init__(self, ("", port), HTTPHandler)
except:
self.address_family = socket.AF_INET
BaseHTTPServer.HTTPServer.__init__(self, ("", port), HTTPHandler)
threading.Thread.__init__(self, name="HTTPThread")
self.host = host
self.port = port
if context:
self.context = context
if not self.context.startswith("/"):
self.context = "/" + self.context
if not self.context.endswith("/"):
self.context += "/"
else:
self.context = "/"
self.docroot = docroot
if index:
self.index = index
else:
self.index = "index.html"
self.handler = handler
self.auth = auth
if (realm == None):
self.authenticateHeader = "Basic realm=webiopi"
else:
self.authenticateHeader = "Basic realm=%s" % realm
self.running = True
self.start()
def get_token_to_login(self, port=8000):
# WARNING: This method is NOT thread-safe
# FIXME: find another way to store token and ensure method works properly
class AllegroRequestHandler(BaseHTTPRequestHandler):
user_token = None
def do_GET(self):
token = self.path.rsplit('?code=', 1)[-1]
AllegroRequestHandler.user_token = token
server = HTTPServer(('0.0.0.0', port), AllegroRequestHandler)
server.handle_request()
user_token = AllegroRequestHandler.user_token
AllegroRequestHandler.user_token = None
return user_token
def __init__(self):
self.host = "localhost"
self.port = next_free_port()
self.url = "http://{}:{}".format(self.host, self.port)
self.hook = error_hook
# Invoke parent class (HTTPServer) constructor
super().__init__((self.host, self.port), MockHTTPRequestHandler)
def __init__(self):
self.log = logging.getLogger('%s.%s' % (self.__class__.__module__, self.__class__.__name__))
self.local_addr = self.listen_port()
self.log.info('Creating HTTP server at %s', self.local_addr)
http_server.HTTPServer.__init__(self, self.local_addr, OAuthTokenHTTPHandler)
self.oauth_verifier = None
def serve(host, port):
"""Serve HTTP requests."""
address = (host, port)
server = HTTPServer(address, RequestHandler)
print('Listening for HTTP requests on {}:{:d} ...'.format(*address))
server.serve_forever()
def __init__(self):
self.port = _find_available_port()
self.host = "localhost"
self.server = HTTPServer((self.host, self.port), ListenerHandler)
self.listening_thread = Thread(target=self._start_listening)
self.listening_thread.daemon = True
self.listening_thread.name = "mock-logzio-listener"
self.listening_thread.start()
self.logs_list = logs_list.list
self.persistent_flags = persistent_flags
def setup_client(handler):
host = 'localhost'
port = get_free_port()
print port
server = HTTPServer((host, port), handler)
server_thread = Thread(target=server.serve_forever)
server_thread.setDaemon(True)
server_thread.start()
client = alluxio.Client(host, port, timeout=60)
return client, lambda: server.shutdown
def StartHelpServer():
global HelpServ
HelpServ = HTTPServer((SrvHost,HlpPort),HelpHandler)
HelpServ.serve_forever()
################################################################################
# Application Handler
################################################################################
# Application Handler
def start():
httpd = BaseHTTPServer.HTTPServer(("", PORT), Handler)
print("serving at port {}".format(PORT))
try:
httpd.serve_forever()
except KeyboardInterrupt:
print(" shutting down...")
httpd.shutdown()
def start(self):
""" Prepares request handler and starts web server """
handler = self.prepare_request_handler()
host = self.get_ip()
port = self.config[WEB_SERVER][HTTP_PORT]
self.web_server = HTTPServer((host, int(port)), handler)
logging.debug("Web Server Started at %s:%s", host, port)
try:
self.web_server.serve_forever()
except:
pass
def run_server(config, tokenConfig, args):
server_address = (args.address, args.port)
HTTPRequestHandler.config = config
HTTPRequestHandler.tokenConfig = tokenConfig
HTTPRequestHandler.args = args
httpd = HTTPServer(server_address, HTTPRequestHandler)
httpd.serve_forever()
def start_server(test_n, test_e, test_id):
"""
Man this is ugly. But you have to set global variables because it's
impossible to send arguments to the HTTPServer, since you pass the HTTPServer
as a type, not an instance.
:param long int test_n: N for an RSA key
:param long int test_e: E for an RSA key
:param str test_id: Key ID for the test key
"""
global TEST_N
global TEST_E
global TEST_ID
global THREAD
global HTTPD
TEST_N = test_n
TEST_E = test_e
TEST_ID = test_id
server_address = ('127.0.0.1', 0)
HTTPD = HTTPServer(server_address, OauthRequestHandler)
THREAD = threading.Thread(target=HTTPD.serve_forever)
THREAD.daemon = True
THREAD.start()
return HTTPD.server_address
def run(port=8000):
server_address = ('', port)
httpd = HTTPServer(server_address, MyHandler)
print('Starting HTTP server on port {}...'.format(port))
httpd.serve_forever()
def __init__(self , host = '', port = 80):
self._routes = defaultdict(list)
self._server = HTTPServer((host, port), Request)
super().__init__()
self.daemon = True
def run(
server_class=HTTPServer,
handler_class=BaseHTTPRequestHandler, port=8098):
server_address = ('127.0.0.1', port)
httpd = server_class(server_address, handler_class)
try:
httpd.serve_forever()
except KeyboardInterrupt:
httpd.socket.close()
def run():
print('starting server...')
# Server settings
# Choose port 8080, for port 80, which is normally used for a http server, you need root access
server_address = ('0.0.0.0', 8081)
httpd = HTTPServer(server_address, testHTTPServer_RequestHandler)
print('running server...')
httpd.serve_forever()
def main():
server_address = (
os.environ.get('NIXBORG_RECEIVER_ADDRESS', '127.0.0.1'),
int(os.environ.get('NIXBORG_RECEIVER_PORT', 7000))
)
httpd = HTTPServer(server_address, HydraJobsetManagerHandler)
httpd.serve_forever()
def start_server(port, cmd_helper: CmdHelper) -> threading.Thread:
def _thread_func():
_Handler.cmd_helper = cmd_helper
server = HTTPServer(('', port), _Handler)
server.serve_forever()
server.socket.close()
t = threading.Thread(target=_thread_func, daemon=True)
_LOG.info('Starting health server on port {}'.format(port))
t.start()
return t
def startServer():
server = HTTPServer(("localhost", 80), RequestHandler)
print(time.asctime(), "Server started on port 80")
try:
server.serve_forever()
except KeyboardInterrupt:
print("Server stopped...")
