python类run()的实例源码

def run(self, handler): # pragma: no cover
        import fapws._evwsgi as evwsgi
        from fapws import base, config
        port = self.port
        if float(config.SERVER_IDENT[-2:]) > 0.4:
            # fapws3 silently changed its API in 0.5
            port = str(port)
        evwsgi.start(self.host, port)
        # fapws3 never releases the GIL. Complain upstream. I tried. No luck.
        if 'BOTTLE_CHILD' in os.environ and not self.quiet:
            _stderr("WARNING: Auto-reloading does not work with Fapws3.\n")
            _stderr("         (Fapws3 breaks python thread support)\n")
        evwsgi.set_base_module(base)
        def app(environ, start_response):
            environ['wsgi.multiprocess'] = False
            return handler(environ, start_response)
        evwsgi.wsgi_cb(('', app))
        evwsgi.run()

def run(self):
        exists = os.path.exists
        mtime = lambda p: os.stat(p).st_mtime
        files = dict()

        for module in list(sys.modules.values()):
            path = getattr(module, '__file__', '')
            if path[-4:] in ('.pyo', '.pyc'): path = path[:-1]
            if path and exists(path): files[path] = mtime(path)

        while not self.status:
            if not exists(self.lockfile)\
            or mtime(self.lockfile) < time.time() - self.interval - 5:
                self.status = 'error'
                thread.interrupt_main()
            for path, lmtime in list(files.items()):
                if not exists(path) or mtime(path) > lmtime:
                    self.status = 'reload'
                    thread.interrupt_main()
                    break
            time.sleep(self.interval)

def run(self, handler):  # pragma: no cover
        from cherrypy import wsgiserver
        self.options['bind_addr'] = (self.host, self.port)
        self.options['wsgi_app'] = handler

        certfile = self.options.get('certfile')
        if certfile:
            del self.options['certfile']
        keyfile = self.options.get('keyfile')
        if keyfile:
            del self.options['keyfile']

        server = wsgiserver.CherryPyWSGIServer(**self.options)
        if certfile:
            server.ssl_certificate = certfile
        if keyfile:
            server.ssl_private_key = keyfile

        try:
            server.start()
        finally:
            server.stop()

def run(self, handler):  # pragma: no cover
        import fapws._evwsgi as evwsgi
        from fapws import base, config
        port = self.port
        if float(config.SERVER_IDENT[-2:]) > 0.4:
            # fapws3 silently changed its API in 0.5
            port = str(port)
        evwsgi.start(self.host, port)
        # fapws3 never releases the GIL. Complain upstream. I tried. No luck.
        if 'BOTTLE_CHILD' in os.environ and not self.quiet:
            _stderr("WARNING: Auto-reloading does not work with Fapws3.\n")
            _stderr("         (Fapws3 breaks python thread support)\n")
        evwsgi.set_base_module(base)

        def app(environ, start_response):
            environ['wsgi.multiprocess'] = False
            return handler(environ, start_response)

        evwsgi.wsgi_cb(('', app))
        evwsgi.run()

def run(self, handler):
        from eventlet import wsgi, listen, patcher
        if not patcher.is_monkey_patched(os):
            msg = "Bottle requires eventlet.monkey_patch() (before import)"
            raise RuntimeError(msg)
        socket_args = {}
        for arg in ('backlog', 'family'):
            try:
                socket_args[arg] = self.options.pop(arg)
            except KeyError:
                pass
        address = (self.host, self.port)
        try:
            wsgi.server(listen(address, **socket_args), handler,
                        log_output=(not self.quiet))
        except TypeError:
            # Fallback, if we have old version of eventlet
            wsgi.server(listen(address), handler)

def run(self, handler):
        import asyncio
        from aiohttp.wsgi import WSGIServerHttpProtocol
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)

        protocol_factory = lambda: WSGIServerHttpProtocol(
            handler,
            readpayload=True,
            debug=(not self.quiet))
        self.loop.run_until_complete(self.loop.create_server(protocol_factory,
                                                             self.host,
                                                             self.port))

        if 'BOTTLE_CHILD' in os.environ:
            import signal
            signal.signal(signal.SIGINT, lambda s, f: self.loop.stop())

        try:
            self.loop.run_forever()
        except KeyboardInterrupt:
            self.loop.stop()

def run(self):
        exists = os.path.exists
        mtime = lambda p: os.stat(p).st_mtime
        files = dict()

        for module in list(sys.modules.values()):
            path = getattr(module, '__file__', '')
            if path[-4:] in ('.pyo', '.pyc'): path = path[:-1]
            if path and exists(path): files[path] = mtime(path)

        while not self.status:
            if not exists(self.lockfile)\
            or mtime(self.lockfile) < time.time() - self.interval - 5:
                self.status = 'error'
                thread.interrupt_main()
            for path, lmtime in list(files.items()):
                if not exists(path) or mtime(path) > lmtime:
                    self.status = 'reload'
                    thread.interrupt_main()
                    break
            time.sleep(self.interval)

def run(servername, ip, port, softcron=True, logging=False, profiler=None,
        options=None):
    if servername == 'gevent':
        from gevent import monkey
        monkey.patch_all()
    elif servername == 'eventlet':
        import eventlet
        eventlet.monkey_patch()

    import gluon.main

    if logging:
        application = gluon.main.appfactory(wsgiapp=gluon.main.wsgibase,
                                            logfilename='httpserver.log',
                                            profiler_dir=profiler)
    else:
        application = gluon.main.wsgibase
    if softcron:
        from gluon.settings import global_settings
        global_settings.web2py_crontype = 'soft'
    getattr(Servers, servername)(application, (ip, int(port)), options=options)

def run():
    if not watch:
        exit(1)

    if before_command:
        click.echo('==> reloopd, INFO  : running RELOOP_BEFORE_CMD')
        subprocess.call(shlex.split(before_command))

    if not command:
        click.echo('ERROR: environment variable RELOOP_CMD is not set! Exiting.')
        exit(1)

    global proc
    proc = subprocess.Popen(shlex.split(command))

    click.echo('==> reloopd, INFO  : watching {0} {1})'.format(('directory' if os.path.isdir(watch) else 'file'),
                                                               os.path.abspath(watch)))

    notifier = inotify.INotify()
    notifier.startReading()
    # recursive=True causes this whole thing to barely work... no FS changes will be detected.
    notifier.watch(filepath.FilePath(str(os.path.abspath(watch))), autoAdd=True, callbacks=[on_change], recursive=True)
    reactor.run()

def main():
    with open(sys.argv[1]) as configuration_file:
        configuration = json.load(configuration_file)

    if len(sys.argv) == 3:
        test_daemon_client = xmlrpclib.ServerProxy("http://{}".format(sys.argv[2]))
        callback = partial(onUpdateFunction, test_daemon_client)
    else:
        callback = partial(noop)

    class ReplicatedValue(ExponentialBackoffResolutionStrategyMixin, SimpleSynchronizationStrategyMixin,
                          BaseReplicatedValue):
        '''
        Mixes just the resolution and synchronization strategies into the base class
        '''

    id = configuration["host"]["id"]
    state_file = "./paxos_state_{}.json".format(id)
    peers = {peer["id"]: (peer["address"], peer["port"]) for peer in configuration["peers"]}
    peers[id] = (configuration["host"]["address"], configuration["host"]["port"])
    r = ReplicatedValue(id, peers.keys(), state_file, callback)
    m = Messenger(id, peers, r)
    if len(sys.argv) == 3:
        listener = TestDaemonListener(m)
    reactor.run()

def run(self, **kwargs):
        """ Calls :func:`run` with the same parameters. """
        run(self, **kwargs)

def run(self, handler):  # pragma: no cover
        pass

def run(self, handler):  # pragma: no cover
        from wsgiref.handlers import CGIHandler

        def fixed_environ(environ, start_response):
            environ.setdefault('PATH_INFO', '')
            return handler(environ, start_response)

        CGIHandler().run(fixed_environ)

def run(self, handler):  # pragma: no cover
        import flup.server.fcgi
        self.options.setdefault('bindAddress', (self.host, self.port))
        flup.server.fcgi.WSGIServer(handler, **self.options).run()

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

def run(self, handler):
        from waitress import serve
        serve(handler, host=self.host, port=self.port, _quiet=self.quiet, **self.options)

def run(self, handler):  # pragma: no cover
        from paste import httpserver
        from paste.translogger import TransLogger
        handler = TransLogger(handler, setup_console_handler=(not self.quiet))
        httpserver.serve(handler,
                         host=self.host,
                         port=str(self.port), **self.options)

def run(self, handler):
        from meinheld import server
        server.listen((self.host, self.port))
        server.run(handler)

def run(self, handler):  # pragma: no cover
        import tornado.wsgi, tornado.httpserver, tornado.ioloop
        container = tornado.wsgi.WSGIContainer(handler)
        server = tornado.httpserver.HTTPServer(container)
        server.listen(port=self.port, address=self.host)
        tornado.ioloop.IOLoop.instance().start()