/usr/lib/python2.7/dist-packages/livereload/handlers.py is in python-livereload 2.5.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | # -*- coding: utf-8 -*-
"""
livereload.handlers
~~~~~~~~~~~~~~~~~~~
HTTP and WebSocket handlers for livereload.
:copyright: (c) 2013 - 2015 by Hsiaoming Yang
:license: BSD, see LICENSE for more details.
"""
import os
import time
import logging
from pkg_resources import resource_string
from tornado import web
from tornado import ioloop
from tornado import escape
from tornado.websocket import WebSocketHandler
from tornado.util import ObjectDict
logger = logging.getLogger('livereload')
class LiveReloadHandler(WebSocketHandler):
waiters = set()
watcher = None
live_css = None
_last_reload_time = None
def allow_draft76(self):
return True
def check_origin(self, origin):
return True
def on_close(self):
if self in LiveReloadHandler.waiters:
LiveReloadHandler.waiters.remove(self)
def send_message(self, message):
if isinstance(message, dict):
message = escape.json_encode(message)
try:
self.write_message(message)
except:
logger.error('Error sending message', exc_info=True)
@classmethod
def start_tasks(cls):
if cls._last_reload_time:
return
if not cls.watcher._tasks:
logger.info('Watch current working directory')
cls.watcher.watch(os.getcwd())
cls._last_reload_time = time.time()
logger.info('Start watching changes')
if not cls.watcher.start(cls.poll_tasks):
logger.info('Start detecting changes')
ioloop.PeriodicCallback(cls.poll_tasks, 800).start()
@classmethod
def poll_tasks(cls):
filepath, delay = cls.watcher.examine()
if not filepath or delay == 'forever' or not cls.waiters:
return
reload_time = 3
if delay:
reload_time = max(3 - delay, 1)
if filepath == '__livereload__':
reload_time = 0
if time.time() - cls._last_reload_time < reload_time:
# if you changed lot of files in one time
# it will refresh too many times
logger.info('Ignore: %s', filepath)
return
if delay:
loop = ioloop.IOLoop.current()
loop.call_later(delay, cls.reload_waiters)
else:
cls.reload_waiters()
@classmethod
def reload_waiters(cls, path=None):
logger.info(
'Reload %s waiters: %s',
len(cls.waiters),
cls.watcher.filepath,
)
if path is None:
path = cls.watcher.filepath or '*'
msg = {
'command': 'reload',
'path': path,
'liveCSS': cls.live_css,
'liveImg': True,
}
cls._last_reload_time = time.time()
for waiter in cls.waiters:
try:
waiter.write_message(msg)
except:
logger.error('Error sending message', exc_info=True)
cls.waiters.remove(waiter)
def on_message(self, message):
"""Handshake with livereload.js
1. client send 'hello'
2. server reply 'hello'
3. client send 'info'
"""
message = ObjectDict(escape.json_decode(message))
if message.command == 'hello':
handshake = {
'command': 'hello',
'protocols': [
'http://livereload.com/protocols/official-7',
],
'serverName': 'livereload-tornado',
}
self.send_message(handshake)
if message.command == 'info' and 'url' in message:
logger.info('Browser Connected: %s' % message.url)
LiveReloadHandler.waiters.add(self)
class LiveReloadJSHandler(web.RequestHandler):
def get(self):
self.set_header('Content-Type', 'application/javascript')
self.write(resource_string(__name__, 'vendors/livereload.js'))
class ForceReloadHandler(web.RequestHandler):
def get(self):
path = self.get_argument('path', default=None) or '*'
LiveReloadHandler.reload_waiters(path)
self.write('ok')
class StaticFileHandler(web.StaticFileHandler):
def should_return_304(self):
return False
|