super basic watchman support

CHANGES.rst

@@ -1,6 +1,10 @@
 unreleased
 ==========
 
+- Added watchman support via ``hupper.watchman.WatchmanFileMonitor``.
+  This is the new preferred file monitor on systems supporting unix sockets.
+  See https://github.com/Pylons/hupper/pull/32
+
 - The ``hupper.watchdog.WatchdogFileMonitor`` will now output some info
   when it receives ulimit or other errors from ``watchdog``.
   See https://github.com/Pylons/hupper/pull/33

docs/api.rst

@@ -12,6 +12,8 @@
 
   .. autofunction:: is_watchdog_supported
 
+  .. autofunction:: is_watchman_supported
+
 .. automodule:: hupper.reloader
 
   .. autoclass:: Reloader
@@ -34,9 +36,11 @@
 .. automodule:: hupper.polling
 
   .. autoclass:: PollingFileMonitor
-     :members:
 
 .. automodule:: hupper.watchdog
 
   .. autoclass:: WatchdogFileMonitor
-     :members:
+
+.. automodule:: hupper.watchman
+
+  .. autoclass:: WatchmanFileMonitor

docs/index.rst

@@ -4,9 +4,15 @@ hupper
 
 ``hupper`` is monitor for your Python process. When files change, the process
 will be restarted. It can be extended to watch arbitrary files. Reloads can
-also be triggered manually from code. File monitoring can be done using
-basic polling or using inotify-style filesystem events if watchdog_ is
-installed.
+also be triggered manually from code.
+
+Builtin file monitors (in order of preference):
+
+- :ref:`watchman_support`
+
+- :ref:`watchdog_support`
+
+- :ref:`polling_support`
 
 Installation
 ============
@@ -44,8 +50,29 @@ Once you have a copy of the source, you can install it with:
 
 .. _Github repo: https://github.com/Pylons/hupper
 
-Watchdog support
-----------------
+Builtin File Monitors
+=====================
+
+.. _watchman_support:
+
+Watchman
+--------
+
+If the `watchman <https://facebook.github.io/watchman/>`_ daemon is running,
+it is the preferred mechanism for monitoring files.
+
+On MacOS it can be installed via:
+
+.. code-block:: console
+
+    $ brew install watchman
+
+Implementation: :class:`hupper.watchman.WatchmanFileMonitor`
+
+.. _watchdog_support:
+
+Watchdog
+--------
 
 If `watchdog <https://pypi.org/project/watchdog/>`_ is installed, it will be
 used to more efficiently watch for changes to files.
@@ -57,6 +84,20 @@ used to more efficiently watch for changes to files.
 This is an optional dependency and if it's not installed, then ``hupper`` will
 fallback to less efficient polling of the filesystem.
 
+Implementation: :class:`hupper.watchdog.WatchdogFileMonitor`
+
+.. _polling_support:
+
+Polling
+-------
+
+The least efficient but most portal approach is to use basic file polling.
+
+The ``reload_interval`` parameter controls how often the filesystem is scanned
+and defaults to once per second.
+
+Implementation: :class:`hupper.polling.PollingFileMonitor`
+
 Command-line Usage
 ==================
 

src/hupper/__init__.py

@@ -2,6 +2,7 @@
 # flake8: noqa
 
 from .utils import is_watchdog_supported
+from .utils import is_watchman_supported
 
 from .reloader import (
     start_reloader,

src/hupper/reloader.py

@@ -10,8 +10,11 @@
 from .compat import queue
 from .ipc import ProcessGroup
 from .logger import DefaultLogger
-from .utils import resolve_spec
-from .utils import is_watchdog_supported
+from .utils import (
+    resolve_spec,
+    is_watchdog_supported,
+    is_watchman_supported,
+)
 from .worker import (
     Worker,
     is_active,
@@ -230,6 +233,11 @@ def find_default_monitor_factory(logger):
 
         logger.debug('File monitor backend: ' + spec)
 
+    elif is_watchman_supported():
+        from .watchman import WatchmanFileMonitor as monitor_factory
+
+        logger.debug('File monitor backend: watchman')
+
     elif is_watchdog_supported():
         from .watchdog import WatchdogFileMonitor as monitor_factory
 

src/hupper/utils.py

@@ -1,4 +1,9 @@
 import importlib
+import json
+import os
+import subprocess
+
+from .compat import WIN
 
 
 def resolve_spec(spec):
@@ -15,3 +20,28 @@ def is_watchdog_supported():
     except ImportError:
         return False
     return True
+
+
+def is_watchman_supported():
+    """ Return ``True`` if watchman is available."""
+    if WIN:
+        # for now we aren't bothering with windows sockets
+        return False
+
+    try:
+        sockpath = get_watchman_sockpath()
+        return bool(sockpath)
+    except Exception:
+        return False
+
+
+def get_watchman_sockpath(binpath='watchman'):
+    """ Find the watchman socket or raise."""
+    path = os.getenv('WATCHMAN_SOCK')
+    if path:
+        return path
+
+    cmd = [binpath, '--output-encoding=json', 'get-sockname']
+    result = subprocess.check_output(cmd)
+    result = json.loads(result)
+    return result['sockname']

src/hupper/watchman.py

@@ -0,0 +1,141 @@
+# check ``hupper.utils.is_watchman_supported`` before using this module
+import json
+import os
+import socket
+import threading
+import time
+
+from .compat import PY2
+from .interfaces import IFileMonitor
+from .utils import get_watchman_sockpath
+
+
+class WatchmanFileMonitor(threading.Thread, IFileMonitor):
+    """
+    An :class:`hupper.interfaces.IFileMonitor` that uses Facebook's
+    ``watchman`` daemon to detect changes.
+
+    ``callback`` is a callable that accepts a path to a changed file.
+
+    """
+    def __init__(
+        self,
+        callback,
+        logger,
+        sockpath=None,
+        binpath='watchman',
+        timeout=1.0,
+        **kw
+    ):
+        super(WatchmanFileMonitor, self).__init__()
+        self.callback = callback
+        self.logger = logger
+        self.paths = set()
+        self.dirpaths = set()
+        self.lock = threading.Lock()
+        self.enabled = True
+        self.sockpath = sockpath
+        self.binpath = binpath
+        self.timeout = timeout
+
+    def add_path(self, path):
+        with self.lock:
+            dirpath = os.path.dirname(path)
+            if dirpath not in self.dirpaths:
+                self._schedule(dirpath)
+                self.dirpaths.add(dirpath)
+
+            if path not in self.paths:
+                self.paths.add(path)
+
+    def start(self):
+        sockpath = self._resolve_sockpath()
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        sock.settimeout(self.timeout)
+        sock.connect(sockpath)
+        self._sock = sock
+        self._recvbufs = []
+
+        self._send(['version'])
+        result = self._recv()
+        self.logger.debug('Connected to watchman v' + result['version'] + '.')
+
+        super(WatchmanFileMonitor, self).start()
+
+    def join(self):
+        try:
+            return super(WatchmanFileMonitor, self).join()
+        finally:
+            self._sock.close()
+            self._sock = None
+
+    def run(self):
+        while self.enabled:
+            try:
+                result = self._recv()
+                if 'error' in result:
+                    self.logger.error('watchman error=' + result['error'])
+                elif 'subscription' in result:
+                    root = result['root']
+                    files = result['files']
+                    with self.lock:
+                        for f in files:
+                            path = os.path.join(root, f)
+                            if path in self.paths:
+                                self.callback(path)
+            except socket.timeout:
+                pass
+
+    def stop(self):
+        self.enabled = False
+
+    def _resolve_sockpath(self):
+        if self.sockpath:
+            return self.sockpath
+        return get_watchman_sockpath(self.binpath)
+
+    def _schedule(self, dirpath):
+        self._send([
+            'subscribe',
+            dirpath,
+            dirpath,
+            {
+                # +1 second because we don't want any buffered changes
+                # if the daemon is already watching the folder
+                'since': int(time.time() + 1),
+                'expression': [
+                    'type', 'f',
+                ],
+                'fields': ['name'],
+            },
+        ])
+
+    def _readline(self):
+        # buffer may already have a line
+        if len(self._recvbufs) == 1 and b'\n' in self._recvbufs[0]:
+            line, b = self._recvbufs[0].split(b'\n', 1)
+            self._recvbufs = [b]
+            return line
+
+        while True:
+            b = self._sock.recv(4096)
+            if not b:
+                raise RuntimeError('lost connection to watchman')
+            if b'\n' in b:
+                result = b''.join(self._recvbufs)
+                line, b = b.split(b'\n', 1)
+                self.buf = [b]
+                return result + line
+            self._recvbufs.append(b)
+
+    def _recv(self):
+        line = self._readline()
+        if not PY2:
+            line = line.decode('utf8')
+        return json.loads(line)
+
+    def _send(self, msg):
+        cmd = json.dumps(msg)
+        if not PY2:
+            cmd = cmd.encode('ascii')
+        self._sock.sendall(cmd + b'\n')

src/hupper/worker.py

@@ -190,6 +190,7 @@ def __init__(self, pipe):
         self.pipe = pipe
 
     def watch_files(self, files):
+        files = [os.path.abspath(f) for f in files]
         self.pipe.send(('watch', files))
 
     def trigger_reload(self):

tests/myapp/cli.py

@@ -13,6 +13,7 @@ def parse_options(args):
     parser.add_argument('--callback-file')
     parser.add_argument('--watch-file', action='append', dest='watch_files',
                         default=[])
+    parser.add_argument('--watchman', action='store_true')
     parser.add_argument('--watchdog', action='store_true')
     parser.add_argument('--poll', action='store_true')
     parser.add_argument('--poll-interval', type=int)
@@ -37,6 +38,10 @@ def main(args=None):
             from hupper.watchdog import WatchdogFileMonitor
             kw['monitor_factory'] = WatchdogFileMonitor
 
+        if opts.watchman:
+            from hupper.watchman import WatchmanFileMonitor
+            kw['monitor_factory'] = WatchmanFileMonitor
+
         if opts.reload_interval:
             kw['reload_interval'] = opts.reload_interval