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()
python类port()的实例源码
def start_multiproxy():
import multiproxy
class ProxyRequestHandler(multiproxy.RequestHandler):
def get_relay_addr_uri(self, pipeline, header_list):
""" Returns relay host, port.
May modify self.request_uri or header list (excluding the first element)
Raises exception if connection not allowed.
"""
comps = self.request_uri.split('/')
if len(comps) > 1 and comps[1] and comps[1] in Options['site_list']:
# Site server
site_number = 1+Options['site_list'].index(comps[1])
retval = SiteProps.relay_map(site_number)
elif Global.relay_forward and not self.request_uri.startswith('/_'):
# Not URL starting with '_'; forward to underlying website
retval = Global.relay_forward
else:
# Root server
retval = SiteProps.relay_map(0)
print >> sys.stderr, 'ABC: get_relay_addr_uri:', sliauth.iso_date(nosubsec=True), self.ip_addr, self.request_uri, retval
return retval
Global.proxy_server = multiproxy.ProxyServer(Options['host'], Options['port'], ProxyRequestHandler, log_interval=0,
io_loop=IOLoop.current(), xheaders=True, masquerade="server/1.2345", ssl_options=Options['ssl_options'], debug=True)
def main():
''' main ??
'''
# ?? search_engin_server
ioloop = tornado.ioloop.IOLoop.instance()
server = tornado.httpserver.HTTPServer(Application(), xheaders=True)
server.listen(options.port)
def sig_handler(sig, _):
''' ??????
'''
logging.warn("Caught signal: %s", sig)
shutdown(ioloop, server)
signal.signal(signal.SIGTERM, sig_handler)
signal.signal(signal.SIGINT, sig_handler)
ioloop.start()
def main(first):
#if os.fork() != 0:
# exit()
if str(first) == 'duo':
freeze_support()
print("Quit the server with CONTROL-C.")
app = application
http_server = tornado.httpserver.HTTPServer(app)
http_server.bind(options.port)
http_server.start(num_processes = 8)
tornado.ioloop.IOLoop.instance().start()
elif str(first) == 'dan':
app = application
app.listen(options.port)
print ("Starting development server at http://172.29.152.3:" + str(options.port) )
print ("Quit the server with CONTROL-C.")
tornado.ioloop.IOLoop.instance().start()
else:
print ("error command: duo or dan?")
def main():
base_dir = os.path.dirname(__file__)
settings = {
"cookie_secret": config.settings.board['cookie_secret'],
"template_path":os.path.join(base_dir,"template"),
"static_path":os.path.join(base_dir,"static"),
"thumbnail_path":os.path.join(base_dir,"thumbnail"),
"debug":True,
"xsrf_cookies":True,
}
application = tornado.web.Application([
tornado.web.url(r"/",MainHandler,name="main"),
tornado.web.url(r"/db_schedule",Get_DB_Data),
],**settings)
http_server = tornado.httpserver.HTTPServer(application)
http_server.listen(options.port)
tornado.ioloop.IOLoop.instance().start()
def main():
parse_command_line(final=False)
parse_config_file(options.config_file)
app = Application(
[
('/', MainHandler),
('/login', LoginHandler),
('/logout', LogoutHandler),
],
login_url='/login',
**options.group_dict('application'))
app.listen(options.port)
logging.info('Listening on http://localhost:%d' % options.port)
IOLoop.instance().start()
def main():
parse_command_line()
app = Application([('/', ChunkHandler)])
app.listen(options.port, address='127.0.0.1')
def callback(response):
response.rethrow()
assert len(response.body) == (options.num_chunks * options.chunk_size)
logging.warning("fetch completed in %s seconds", response.request_time)
IOLoop.instance().stop()
logging.warning("Starting fetch with curl client")
curl_client = CurlAsyncHTTPClient()
curl_client.fetch('http://localhost:%d/' % options.port,
callback=callback)
IOLoop.instance().start()
logging.warning("Starting fetch with simple client")
simple_client = SimpleAsyncHTTPClient()
simple_client.fetch('http://localhost:%d/' % options.port,
callback=callback)
IOLoop.instance().start()
def main():
parse_command_line(final=False)
parse_config_file(options.config_file)
app = Application(
[
('/', MainHandler),
('/login', LoginHandler),
('/logout', LogoutHandler),
],
login_url='/login',
**options.group_dict('application'))
app.listen(options.port)
logging.info('Listening on http://localhost:%d' % options.port)
IOLoop.current().start()
def main():
parse_command_line()
app = Application([('/', ChunkHandler)])
app.listen(options.port, address='127.0.0.1')
def callback(response):
response.rethrow()
assert len(response.body) == (options.num_chunks * options.chunk_size)
logging.warning("fetch completed in %s seconds", response.request_time)
IOLoop.current().stop()
logging.warning("Starting fetch with curl client")
curl_client = CurlAsyncHTTPClient()
curl_client.fetch('http://localhost:%d/' % options.port,
callback=callback)
IOLoop.current().start()
logging.warning("Starting fetch with simple client")
simple_client = SimpleAsyncHTTPClient()
simple_client.fetch('http://localhost:%d/' % options.port,
callback=callback)
IOLoop.current().start()
def main():
parse_command_line()
app = tornado.web.Application(
[
(r"/", MainHandler),
(r"/a/message/new", MessageNewHandler),
(r"/a/message/updates", MessageUpdatesHandler),
],
cookie_secret="__TODO:_GENERATE_YOUR_OWN_RANDOM_VALUE_HERE__",
template_path=os.path.join(os.path.dirname(__file__), "templates"),
static_path=os.path.join(os.path.dirname(__file__), "static"),
xsrf_cookies=True,
debug=options.debug,
)
app.listen(options.port)
tornado.ioloop.IOLoop.current().start()
def make_server(application, conf_dir=None):
"""
Configure the server return the server instance
"""
if conf_dir:
load_config(conf_dir)
configure_syslog()
log_config()
if options.use_ssl:
ssl_options = ssl_server_options()
server = tornado.httpserver.HTTPServer(
application, ssl_options=ssl_options)
general_logger.info(
'start tornado https server at https://%s:%s'
' with ssl_options: %s', options.ip, options.port, ssl_options)
else:
server = tornado.httpserver.HTTPServer(application)
general_logger.info('start tornado http server at http://{0}:{1}'.format(
options.ip, options.port))
server.bind(options.port, options.ip)
return server
def main():
tornado.options.parse_command_line()
app = App()
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
if options.debug:
# autorelaod for template file
tornado.autoreload.start()
for root, dir, files in os.walk(TEMPLATE_PATH):
for item in files:
if item.endswith('.html'):
tornado.autoreload.watch(os.path.join(root, item))
try:
tornado.ioloop.IOLoop.current().start()
except:
tornado.ioloop.IOLoop.current().stop()
def main():
parse_command_line()
app = tornado.web.Application(
[
(r"/", MainHandler),
(r"/a/message/new", MessageNewHandler),
(r"/a/message/updates", MessageUpdatesHandler),
],
cookie_secret="153A3FDSIKM56",
template_path=os.path.join(os.path.dirname(__file__), "templates"),
static_path=os.path.join(os.path.dirname(__file__), "static"),
xsrf_cookies=True,
debug=options.debug,
)
app.listen(options.port)
tornado.ioloop.IOLoop.current().start()
def setup():
system = platform.system()
engine_path = os.path.dirname(__file__)
engine_paths = dict([
('Darwin', 'engines/stockfish/Mac/stockfish-7-64'),
('Linux', 'engines/stockfish/Linux/stockfish-7-x64'),
('Linux2', 'engines/stockfish/Linux/stockfish-7-x64'),
])
if not system in engine_paths:
raise Exception('No engine for OS')
engine_path = os.path.join(engine_path, engine_paths[system])
environment = os.environ['ENVIRONMENT'] if 'ENVIRONMENT' in os.environ else 'PRODUCTION'
port = 5000
debug = True
if environment != 'development':
port = os.environ.get('PORT', 5000)
debug = False
define('port', default=port, help='run on the given port', type=int)
define('debug', default=debug, help='run in debug mode')
define('path-to-engine', default=engine_path, help='the location of the chess engine')
def main():
parse_command_line(final=False)
parse_config_file(options.config_file)
app = Application(
[
('/', MainHandler),
('/login', LoginHandler),
('/logout', LogoutHandler),
],
login_url='/login',
**options.group_dict('application'))
app.listen(options.port)
logging.info('Listening on http://localhost:%d' % options.port)
IOLoop.current().start()
def main():
parse_command_line()
app = Application([('/', ChunkHandler)])
app.listen(options.port, address='127.0.0.1')
def callback(response):
response.rethrow()
assert len(response.body) == (options.num_chunks * options.chunk_size)
logging.warning("fetch completed in %s seconds", response.request_time)
IOLoop.current().stop()
logging.warning("Starting fetch with curl client")
curl_client = CurlAsyncHTTPClient()
curl_client.fetch('http://localhost:%d/' % options.port,
callback=callback)
IOLoop.current().start()
logging.warning("Starting fetch with simple client")
simple_client = SimpleAsyncHTTPClient()
simple_client.fetch('http://localhost:%d/' % options.port,
callback=callback)
IOLoop.current().start()
def main():
parse_command_line()
app = tornado.web.Application(
[
(r"/", MainHandler),
(r"/a/message/new", MessageNewHandler),
(r"/a/message/updates", MessageUpdatesHandler),
],
cookie_secret="__TODO:_GENERATE_YOUR_OWN_RANDOM_VALUE_HERE__",
template_path=os.path.join(os.path.dirname(__file__), "templates"),
static_path=os.path.join(os.path.dirname(__file__), "static"),
xsrf_cookies=True,
debug=options.debug,
)
app.listen(options.port)
tornado.ioloop.IOLoop.current().start()
def main():
parse_command_line()
redis.connect(host=options.redis_host)
app = tornado.web.Application(
[
(r'/', MainHandler),
(r'/oauth', OAuthHandler),
(r'/command', CommandHandler),
(r'/button', ButtonHandler),
],
template_path=os.path.join(os.path.dirname(__file__), 'templates'),
static_path=os.path.join(os.path.dirname(__file__), 'static'),
debug=options.debug,
)
app.listen(options.port)
ioloop = tornado.ioloop.IOLoop.current()
ioloop.start()
def main():
tornado.options.parse_command_line()
print '\ncreating app=Applicatoin()'
app = Application()
print '\napp= Aapplication() created -> app : ', app
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
print "\nStarting server on http://127.0.0.1:%s" % options.port
try:
tornado.ioloop.IOLoop.current().start()
except KeyboardInterrupt:
print '\n\nEXCEPTION KEYBOARDINTERRUPT INITIATED\n'
print "Stopping Server....\n"
print 'closing all websocket connections objects and corresponsding pika client objects\n'
# wsparticipants = apps.main.views.websocketParticipants
# for ws in wsparticipants:
# print '\nCLOSING WS.on_close object : ', ws
# ws.on_close()
# apps.main.views.websocketParticipants = []
print "\nServer Stopped.\n"
def main():
tornado.options.parse_command_line()
app = Application()
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
LOGGER.info('[server.main] Starting server on http://127.0.0.1:%s', options.port)
try:
LOGGER.info("\n[server.main] Server Started.\n")
tornado.ioloop.IOLoop.current().start()
except KeyboardInterrupt:
LOGGER.error('\n[server.main] EXCEPTION KEYBOARDINTERRUPT INITIATED\n')
LOGGER.info("[server.main] Stopping Server....")
LOGGER.info('[server.main] closing all websocket connections objects and corresponsding pika client objects')
LOGGER.info("\n[server.main] Server Stopped.")
def __init__(self):
# register tornado's RequestHandler to designated context path
handlers = [
(r'/publish', PublishHandler),
(r'/subscribe', SubscribeHandler),
(r'/', IndexHandler),
]
settings = dict(
template_path=path.join(path.dirname(__file__), 'templates'),
static_path=path.join(path.dirname(__file__), 'static'),
)
super(Application, self).__init__(handlers, **settings)
# redis singeleton to share among handlers
self.redis = redis.StrictRedis(host=options.redis_host, port=options.redis_port, db=options.redis_db, password=options.redis_password)
# redis pubsub singeleton to share among handler(pubsub is used to subscribe channels)
self.pubsub = self.redis.pubsub(ignore_subscribe_messages=True)
self.pubsub.subscribe(options.redis_channel)
# logger to record log
self.logger = logger
# base handler
def main():
parse_command_line(final=False)
parse_config_file(options.config_file)
app = Application(
[
('/', MainHandler),
('/login', LoginHandler),
('/logout', LogoutHandler),
],
login_url='/login',
**options.group_dict('application'))
app.listen(options.port)
logging.info('Listening on http://localhost:%d' % options.port)
IOLoop.current().start()
def main():
parse_command_line()
app = Application([('/', ChunkHandler)])
app.listen(options.port, address='127.0.0.1')
def callback(response):
response.rethrow()
assert len(response.body) == (options.num_chunks * options.chunk_size)
logging.warning("fetch completed in %s seconds", response.request_time)
IOLoop.current().stop()
logging.warning("Starting fetch with curl client")
curl_client = CurlAsyncHTTPClient()
curl_client.fetch('http://localhost:%d/' % options.port,
callback=callback)
IOLoop.current().start()
logging.warning("Starting fetch with simple client")
simple_client = SimpleAsyncHTTPClient()
simple_client.fetch('http://localhost:%d/' % options.port,
callback=callback)
IOLoop.current().start()
def main():
parse_command_line()
app = tornado.web.Application(
[
(r"/", MainHandler),
(r"/a/message/new", MessageNewHandler),
(r"/a/message/updates", MessageUpdatesHandler),
],
cookie_secret="__TODO:_GENERATE_YOUR_OWN_RANDOM_VALUE_HERE__",
template_path=os.path.join(os.path.dirname(__file__), "templates"),
static_path=os.path.join(os.path.dirname(__file__), "static"),
xsrf_cookies=True,
debug=options.debug,
)
app.listen(options.port)
tornado.ioloop.IOLoop.current().start()
def main():
parse_command_line(final=False)
parse_config_file(options.config_file)
app = Application(
[
('/', MainHandler),
('/login', LoginHandler),
('/logout', LogoutHandler),
],
login_url='/login',
**options.group_dict('application'))
app.listen(options.port)
logging.info('Listening on http://localhost:%d' % options.port)
IOLoop.current().start()
chunk_benchmark.py 文件源码
项目：LinuxBashShellScriptForOps
作者: DingGuodong
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def main():
parse_command_line()
app = Application([('/', ChunkHandler)])
app.listen(options.port, address='127.0.0.1')
def callback(response):
response.rethrow()
assert len(response.body) == (options.num_chunks * options.chunk_size)
logging.warning("fetch completed in %s seconds", response.request_time)
IOLoop.current().stop()
logging.warning("Starting fetch with curl client")
curl_client = CurlAsyncHTTPClient()
curl_client.fetch('http://localhost:%d/' % options.port,
callback=callback)
IOLoop.current().start()
logging.warning("Starting fetch with simple client")
simple_client = SimpleAsyncHTTPClient()
simple_client.fetch('http://localhost:%d/' % options.port,
callback=callback)
IOLoop.current().start()
def main():
parse_command_line()
app = tornado.web.Application(
[
(r"/", MainHandler),
(r"/a/message/new", MessageNewHandler),
(r"/a/message/updates", MessageUpdatesHandler),
],
cookie_secret="__TODO:_GENERATE_YOUR_OWN_RANDOM_VALUE_HERE__",
template_path=os.path.join(os.path.dirname(__file__), "templates"),
static_path=os.path.join(os.path.dirname(__file__), "static"),
xsrf_cookies=True,
debug=options.debug,
)
app.listen(options.port)
tornado.ioloop.IOLoop.current().start()
def main():
tornado.options.parse_command_line()
app = Application()
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
LOGGER.info('[server.main] Starting server on http://127.0.0.1:%s', options.port)
try:
LOGGER.info("\n[server.main] Server Started.\n")
tornado.ioloop.IOLoop.current().start()
except KeyboardInterrupt:
LOGGER.error('\n[server.main] EXCEPTION KEYBOARDINTERRUPT INITIATED\n')
LOGGER.info("[server.main] Stopping Server....")
LOGGER.info('[server.main] closing all websocket connections objects and corresponsding mqtt client objects')
LOGGER.info('Stopping Tornado\'s main iolooop')
# Stopping main thread's ioloop, not to be confused with current thread's ioloop
# which is ioloop.IOLoop.current()
tornado.ioloop.IOLoop.instance().stop()
LOGGER.info("\n[server.main] Server Stopped.")
def main():
tornado.options.parse_command_line()
app = Application()
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
LOGGER.info('[server.main] Starting server on http://127.0.0.1:%s', options.port)
try:
LOGGER.info("\n[server.main] Server Started.\n")
tornado.ioloop.IOLoop.current().start()
except KeyboardInterrupt:
LOGGER.error('\n[server.main] EXCEPTION KEYBOARDINTERRUPT INITIATED\n')
LOGGER.info("[server.main] Stopping Server....")
LOGGER.info('[server.main] closing all websocket connections objects and corresponsding mqtt client objects')
LOGGER.info('Stopping Tornado\'s main iolooop')
# Stopping main thread's ioloop, not to be confused with current thread's ioloop
# which is ioloop.IOLoop.current()
tornado.ioloop.IOLoop.instance().stop()
LOGGER.info("\n[server.main] Server Stopped.")
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
