<s> """<STR_LIT>""" <EOL> import os <EOL> import time <EOL> import logging <EOL> from pkg_resources import resource_string <EOL> from tornado import web <EOL> from tornado import ioloop <EOL> from tornado import escape <EOL> from tornado . websocket import WebSocketHandler <EOL> from tornado . util import ObjectDict <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> class LiveReloadHandler ( WebSocketHandler ) : <EOL> waiters = set ( ) <EOL> watcher = None <EOL> _last_reload_time = None <EOL> def allow_draft76 ( self ) : <EOL> return True <EOL> def check_origin ( self , origin ) : <EOL> return True <EOL> def on_close ( self ) : <EOL> if self in LiveReloadHandler . waiters : <EOL> LiveReloadHandler . waiters . remove ( self ) <EOL> def send_message ( self , message ) : <EOL> if isinstance ( message , dict ) : <EOL> message = escape . json_encode ( message ) <EOL> try : <EOL> self . write_message ( message ) <EOL> except : <EOL> logger . error ( '<STR_LIT>' , exc_info = True ) <EOL> @ classmethod <EOL> def start_tasks ( cls ) : <EOL> if cls . _last_reload_time : <EOL> return <EOL> if not cls . watcher . _tasks : <EOL> logger . info ( '<STR_LIT>' ) <EOL> cls . watcher . watch ( os . getcwd ( ) ) <EOL> cls . _last_reload_time = time . time ( ) <EOL> logger . info ( '<STR_LIT>' ) <EOL> if not cls . watcher . start ( cls . poll_tasks ) : <EOL> logger . info ( '<STR_LIT>' ) <EOL> ioloop . PeriodicCallback ( cls . poll_tasks , <NUM_LIT> ) . start ( ) <EOL> @ classmethod <EOL> def poll_tasks ( cls ) : <EOL> filepath , delay = cls . watcher . examine ( ) <EOL> if not filepath or delay == '<STR_LIT>' or not cls . waiters : <EOL> return <EOL> reload_time = <NUM_LIT:3> <EOL> if delay : <EOL> reload_time = max ( <NUM_LIT:3> - delay , <NUM_LIT:1> ) <EOL> if filepath == '<STR_LIT>' : <EOL> reload_time = <NUM_LIT:0> <EOL> if time . time ( ) - cls . _last_reload_time < reload_time : <EOL> logger . info ( '<STR_LIT>' , filepath ) <EOL> return <EOL> if delay : <EOL> loop = ioloop . IOLoop . current ( ) <EOL> loop . call_later ( delay , cls . reload_waiters ) <EOL> else : <EOL> cls . reload_waiters ( ) <EOL> @ classmethod <EOL> def reload_waiters ( cls , path = None ) : <EOL> logger . info ( <EOL> '<STR_LIT>' , <EOL> len ( cls . waiters ) , <EOL> cls . watcher . filepath , <EOL> ) <EOL> if path is None : <EOL> path = cls . watcher . filepath or '<STR_LIT:*>' <EOL> msg = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:path>' : path , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> } <EOL> cls . _last_reload_time = time . time ( ) <EOL> for waiter in cls . waiters : <EOL> try : <EOL> waiter . write_message ( msg ) <EOL> except : <EOL> logger . error ( '<STR_LIT>' , exc_info = True ) <EOL> cls . waiters . remove ( waiter ) <EOL> def on_message ( self , message ) : <EOL> """<STR_LIT>""" <EOL> message = ObjectDict ( escape . json_decode ( message ) ) <EOL> if message . command == '<STR_LIT:hello>' : <EOL> handshake = { <EOL> '<STR_LIT>' : '<STR_LIT:hello>' , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' , <EOL> ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> self . send_message ( handshake ) <EOL> if message . command == '<STR_LIT:info>' and '<STR_LIT:url>' in message : <EOL> logger . info ( '<STR_LIT>' % message . url ) <EOL> LiveReloadHandler . waiters . add ( self ) <EOL> class LiveReloadJSHandler ( web . RequestHandler ) : <EOL> def get ( self ) : <EOL> self . set_header ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) <EOL> self . write ( resource_string ( __name__ , '<STR_LIT>' ) ) <EOL> class ForceReloadHandler ( web . RequestHandler ) : <EOL> def get ( self ) : <EOL> path = self . get_argument ( '<STR_LIT:path>' , default = None ) or '<STR_LIT:*>' <EOL> LiveReloadHandler . reload_waiters ( path ) <EOL> self . write ( '<STR_LIT>' ) <EOL> class StaticFileHandler ( web . StaticFileHandler ) : <EOL> def should_return_304 ( self ) : <EOL> return False </s>
