python类make_server()的实例源码

ecdsa.py 文件源码 项目:abe-bootstrap 作者: TryCoin-Team 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def serve(store):
    args = store.args
    abe = Aml(store, args)

    if args.host or args.port:
        # HTTP server.
        if args.host is None:
            args.host = "localhost"
        from wsgiref.simple_server import make_server
        port = int(args.port or 80)
        httpd = make_server(args.host, port, abe )
        print "Listening on http://" + args.host + ":" + str(port)
        try:
            httpd.serve_forever()
        except:
            httpd.shutdown()
            raise
test_http_utils.py 文件源码 项目:captaincloud 作者: bpsagar 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def test_make_client(self):
        test = self.Test2(value=10)
        app = bottle_api.make_app(('/api', test))
        server = make_server('localhost', 10001, app)
        process = Thread(
            target=server.serve_forever)
        process.start()

        time.sleep(1)  # Wait for the app to run

        c = client.make_client(
            instance=test, base_url='http://localhost:10001/api')
        data = c.endpoint(arg1='test')
        expected_data = {'works': True, 'arg1': 'test', 'value': 10}
        self.assertEqual(data, expected_data)

        with self.assertRaises(AttributeError):
            c.invalid_endpoint()

        test.endpoint(arg1='test')  # For 100% coverage :D

        app.close()
        server.shutdown()
        process.join()
bottle.py 文件源码 项目:Mmrz-Sync 作者: zhanglintc 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def run(self, app): # pragma: no cover
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        srv.serve_forever()
bottle.py 文件源码 项目:ynm3k 作者: socrateslee 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def run(self, app): # pragma: no cover
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        srv.serve_forever()
bottle.py 文件源码 项目:fgc 作者: mpaulweeks 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def run(self, app): # pragma: no cover
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        srv.serve_forever()
bottle.py 文件源码 项目:base1k 作者: xiaq 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def run(self, app): # pragma: no cover
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        srv.serve_forever()
bottle.py 文件源码 项目:driveboardapp 作者: nortd 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def run(self, app): # pragma: no cover
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        srv.serve_forever()
bottle.py 文件源码 项目:movies-python-py2neo-3.0 作者: neo4j-examples 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def run(self, app): # pragma: no cover
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        srv.serve_forever()
webservice.py 文件源码 项目:octohook 作者: dsnezhkov 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def run(self, handler):
        self.certfile = 'keys/server.pem'

        from wsgiref.simple_server import make_server, WSGIRequestHandler
        import ssl
        if self.quiet:
            class QuietHandler(WSGIRequestHandler):
                def log_request(*args, **kw): pass
            self.options['handler_class'] = QuietHandler
        srv = make_server(self.host, self.port, handler, **self.options)
        try:
            with open(self.certfile):
                pass
            srv.socket = ssl.wrap_socket(
                srv.socket,
                certfile=self.certfile
                )
            srv.serve_forever()
        except IOError as e:
            logging.error("Unable to open Certificate file at location {} {}".
                  format(self.certfile, e))
            raise
tester.py 文件源码 项目:web3.py 作者: ethereum 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def __init__(self, host="127.0.0.1", port=8545, *args, **kwargs):
        if not is_testrpc_available():
            raise Exception("`TestRPCProvider` requires the `eth-testrpc` package to be installed")
        from testrpc.server import get_application

        application = get_application()

        self.server = make_server(
            host,
            port,
            application,
        )

        self.thread = spawn(self.server.serve_forever)
        endpoint_uri = 'http://{0}:{1}'.format(host, port)

        super(TestRPCProvider, self).__init__(endpoint_uri, *args, **kwargs)
bottle.py 文件源码 项目:python-group-proj 作者: Sharcee 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def run(self, app): # pragma: no cover
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        srv.serve_forever()
bottle.py 文件源码 项目:python-group-proj 作者: Sharcee 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def run(self, app): # pragma: no cover
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        srv.serve_forever()
test_authorization_code.py 文件源码 项目:islam-buddy 作者: hamir 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def run_client(queue):
    try:
        app = ClientApplication(
            callback_url="http://127.0.0.1:15487/callback",
            client_id="abc",
            client_secret="xyz",
            provider_url="http://127.0.0.1:15486")

        httpd = make_server('', 15487, app,
                            handler_class=NoLoggingHandler)

        queue.put({"result": 0})

        httpd.serve_forever()
    except Exception as e:
        queue.put({"result": 1, "error_message": str(e)})
conftest.py 文件源码 项目:pyramid_notebook 作者: websauna 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def run(self):
        """
        Open WSGI server to listen to HOST_BASE address
        """
        print("Running")
        parts = urlparse(self.hostbase)
        domain, port = parts.netloc.split(":")
        print(domain, port)
        self.srv = make_server(domain, int(port), self.app)
        try:
            self.srv.serve_forever()
        except:
            # We are a background thread so we have problems to interrupt tests in the case of error
            import traceback
            traceback.print_exc()
            # Failed to start
            self.srv = None
bottle.py 文件源码 项目:Guillotines 作者: P-Monkey 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def run(self, app): # pragma: no cover
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        srv.serve_forever()
fixtures.py 文件源码 项目:schul_cloud_resources_server_tests 作者: schul-cloud 项目源码 文件源码 阅读 29 收藏 0 点赞 0 评论 0
def run(self, app):
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        self.srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        self.srv.serve_forever()
        while not self.get_port(): time.sleep(0.001)
bottle.py 文件源码 项目:Pardus-Bulut 作者: ferhatacikalin 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def run(self, app): # pragma: no cover
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        srv.serve_forever()
webserver.py 文件源码 项目:reahl 作者: reahl 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def __init__(self, config, port):
        super(ReahlWebServer, self).__init__()
        self.in_separate_thread = None
        self.running = False
        self.handlers = {}
        self.httpd_thread = None
        certfile = pkg_resources.resource_filename(__name__, 'reahl_development_cert.pem')
        self.reahl_wsgi_app = WrappedApp(ReahlWSGIApplication(config))
        try:
            https_port = port+363
            self.httpd = ReahlWSGIServer.make_server('', port, self.reahl_wsgi_app)
            self.httpsd = SSLCapableWSGIServer.make_server('', https_port, certfile, self.reahl_wsgi_app)
        except socket.error as ex:
            message = ('Caught socket.error: %s\nThis means that another process is using one of these ports: %s, %s. ' % (ex, port, https_port)) \
                     +'\nIf this happens while running tests, it probably means that a browser client did not close its side of a connection to a previous server you had running - and that the server socket now sits in TIME_WAIT state. Is there perhaps a browser hanging around from a previous run? I have no idea how to fix this automatically... see http://hea-www.harvard.edu/~fine/Tech/addrinuse.html'

            raise AssertionError(message)
KodiServer.py 文件源码 项目:xbmctopython 作者: pybquillast 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def runServer(kodi = '', kodi_home = '', startBrowser=False):
    importer = ksi.KodiScriptImporter(kodi, kodi_home)
    importer.install(True)
    kodiSrv = KodiServer(importer)
    wsgiApp = partial(application, server=kodiSrv)

    server_address = ('localhost', 5000)
    httpd = make_server(
        server_address[0],  # The host name
        server_address[1],  # A port number where to wait for the request
        wsgiApp  # The application object name, in this case a function
    )
    srvThread = threading.Thread(target=httpd.serve_forever)
    if startBrowser:
        webbrowser.open('http://{}:{}'.format(*server_address))
    srvThread.start()
    return httpd
bottle.py 文件源码 项目:installations-sportives-pdl 作者: sebprunier 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def run(self, app): # pragma: no cover
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        from wsgiref.simple_server import make_server
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self): # Prevent reverse DNS lookups please.
                return self.client_address[0]
            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls  = self.options.get('server_class', WSGIServer)

        if ':' in self.host: # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        srv.serve_forever()
bottletlsdaemon.py 文件源码 项目:agent 作者: upd89 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def run(self, handler):
        from wsgiref.simple_server import make_server, WSGIRequestHandler
        import ssl
        if self.quiet:
            class QuietHandler(WSGIRequestHandler):

                def log_request(*args, **kw):
                    pass
            self.options['handler_class'] = QuietHandler
        srv = make_server(self.host, self.port, handler, **self.options)
        srv.socket = ssl.wrap_socket(
            srv.socket,
            keyfile=self.keyfile,
            certfile=self.certfile,
            ca_certs=self.cafile,
            cert_reqs=ssl.CERT_REQUIRED,
            server_side=True)
        srv.serve_forever()
http_server.py 文件源码 项目:vpc-router 作者: romana 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def run(self, handler):
        from wsgiref.simple_server import make_server, WSGIRequestHandler
        if self.quiet:
            class QuietHandler(WSGIRequestHandler):
                def log_request(*args, **kw):
                    pass
            self.options['handler_class'] = QuietHandler
        try:
            self.server = make_server(self.host, self.port, handler,
                                      **self.options)
            self.romana_http.wsgi_server_started = True
            logging.info("HTTP server: Started to listen...")
            self.server.serve_forever()
        except socket.error as e:
            logging.fatal("HTTP server: Cannot open socket "
                          "(error %d: %s)... " %
                          (e.errno, e.strerror))
mfa_slipstream.py 文件源码 项目:mfa_slipstream_poc 作者: decidedlygray 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def run(self, handler):
        from wsgiref.simple_server import make_server, WSGIRequestHandler
        import ssl
        if self.quiet:
            class QuietHandler(WSGIRequestHandler):
                def log_request(*args, **kw): pass
            self.options['handler_class'] = QuietHandler
        srv = make_server(self.host, self.port, handler, **self.options)
        srv.socket = ssl.wrap_socket (
         srv.socket,
         keyfile='mfa_slipstream.pem',
         certfile='mfa_slipstream.pem',  # path to certificate
         server_side=True)
        srv.serve_forever()


##########################################################################
# UTILITY FUNCTIONS
##########################################################################
reports.py 文件源码 项目:nuka 作者: bearstech 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def run_dev_server():  # pragma: no cover
    """Usage: python -m nuka.report <type> <data_filename>"""
    from wsgiref.simple_server import make_server
    engine = nuka.config.get_template_engine()
    type, filename = sys.argv[-2:]

    def app(environ, start_response):
        start_response('200 OK',
                       [('Content-Type', 'text/html; charset=utf8')])
        template = engine.get_template('reports/{0}.html.j2'.format(type))
        with open(filename) as fd:
            data = json.load(fd)
        html = template.render(dict(data=data, dumped_data=json.dumps(data)))
        return [html.encode('utf8')]

    httpd = make_server('', 8000, app)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
cli.py 文件源码 项目:dsq 作者: baverman 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def http(tasks, bind):
    """Http interface using built-in simple wsgi server"""
    from wsgiref.simple_server import make_server
    from .utils import load_manager
    from .http import Application

    host, _, port = bind.partition(':')
    app = Application(load_manager(tasks))
    httpd = make_server(host, int(port), app)
    print('Listen on {}:{} ...'.format(host or '0.0.0.0', port), file=sys.stderr)
    httpd.serve_forever()
bottle.py 文件源码 项目:dabdabrevolution 作者: harryparkdotio 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def run(self, app):  # pragma: no cover
        from wsgiref.simple_server import make_server
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self):  # Prevent reverse DNS lookups please.
                return self.client_address[0]

            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = self.options.get('handler_class', FixedHandler)
        server_cls = self.options.get('server_class', WSGIServer)

        if ':' in self.host:  # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:

                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        self.srv = make_server(self.host, self.port, app, server_cls,
                               handler_cls)
        self.port = self.srv.server_port  # update port actual port (0 means random)
        try:
            self.srv.serve_forever()
        except KeyboardInterrupt:
            self.srv.server_close()  # Prevent ResourceWarning: unclosed socket
            raise
pjf_server.py 文件源码 项目:PyJFuzz 作者: mseclab 项目源码 文件源码 阅读 32 收藏 0 点赞 0 评论 0
def run(self, handler):
        class QuietHandler(WSGIRequestHandler):
            def log_request(*args, **kw):
                pass

            def log_error(self, format, *args):
                pass
        self.options['handler_class'] = QuietHandler
        srv = make_server(self.host, self.port, handler, **self.options)
        srv.serve_forever()
pjf_server.py 文件源码 项目:PyJFuzz 作者: mseclab 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def run(self, handler):
        class QuietHandler(WSGIRequestHandler):
            def log_request(*args, **kw):
                pass

            def log_error(self, format, *args):
                pass
        self.options['handler_class'] = QuietHandler
        srv = make_server(self.host, self.port, handler, **self.options)
        srv.socket = ssl.wrap_socket(srv.socket, certfile=CERT_PATH, server_side=True)
        srv.serve_forever()
server.py 文件源码 项目:kaira 作者: mulonemartin 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def run(self, app):  # pragma: no cover
        from wsgiref.simple_server import make_server
        from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
        import socket

        class FixedHandler(WSGIRequestHandler):
            def address_string(self):  # Prevent reverse DNS lookups please.
                return self.client_address[0]

            def log_request(*args, **kw):
                if not self.quiet:
                    return WSGIRequestHandler.log_request(*args, **kw)

        handler_cls = FixedHandler
        server_cls = WSGIServer

        if ':' in self.host:  # Fix wsgiref for IPv6 addresses.
            if getattr(server_cls, 'address_family') == socket.AF_INET:

                class server_cls(server_cls):
                    address_family = socket.AF_INET6

        self.srv = make_server(self.host, self.port, app, server_cls,
                               handler_cls)
        self.port = self.srv.server_port  # update port actual port (0 means random)
        try:
            self.srv.serve_forever()
        except KeyboardInterrupt:
            self.srv.server_close()  # Prevent ResourceWarning: unclosed socket
            raise
process.py 文件源码 项目:captaincloud 作者: bpsagar 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def run(self):
        """Run the Process"""
        self.runner = TaskRunner()
        self.runner_api = TaskRunnerAPI(process=self, runner=self.runner)

        self.app = bottle_api.make_app(('/api', self.runner_api))
        self.server = make_server(self.host, self.port, self.app)

        self.runner.start()
        self.server.serve_forever()
        self.runner.join()


问题


面经


文章

微信
公众号

扫码关注公众号