inotify: modular architecture for inotify clients

Put the socket init, query generation and response analysis in
a more generic client class.

hgext/inotify/__init__.py

@@ -1,125 +1,127
 # __init__.py - inotify-based status acceleration for Linux
 #
 # Copyright 2006, 2007, 2008 Bryan O'Sullivan <[e-mail]>
 # Copyright 2007, 2008 Brendan Cully <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2, incorporated herein by reference.
 
 '''inotify-based status acceleration for Linux systems
 '''
 
 # todo: socket permissions
 
 from mercurial.i18n import _
 from mercurial import cmdutil, util
-import client, errno, os, server, socket
+import errno, os, server, socket
 from weakref import proxy
+from client import client
 
 def serve(ui, repo, **opts):
     '''start an inotify server for this repository'''
     timeout = opts.get('timeout')
     if timeout:
         timeout = float(timeout) * 1e3
 
     class service:
         def init(self):
             try:
                 self.master = server.master(ui, repo, timeout)
             except server.AlreadyStartedException, inst:
                 raise util.Abort(str(inst))
 
         def run(self):
             try:
                 self.master.run()
             finally:
                 self.master.shutdown()
 
     service = service()
     cmdutil.service(opts, initfn=service.init, runfn=service.run)
 
 def reposetup(ui, repo):
     if not hasattr(repo, 'dirstate'):
         return
 
     # XXX: weakref until hg stops relying on __del__
     repo = proxy(repo)
 
     class inotifydirstate(repo.dirstate.__class__):
         # Set to True if we're the inotify server, so we don't attempt
         # to recurse.
         inotifyserver = False
 
         def status(self, match, ignored, clean, unknown=True):
             files = match.files()
             if '.' in files:
                 files = []
+            cli = client(ui, repo)
             try:
                 if not ignored and not self.inotifyserver:
-                    result = client.query(ui, repo, files, match, False,
-                                          clean, unknown)
+                    result = cli.statusquery(files, match, False,
+                                             clean, unknown)
                     if result and ui.config('inotify', 'debug'):
                         r2 = super(inotifydirstate, self).status(
                             match, False, clean, unknown)
                         for c,a,b in zip('LMARDUIC', result, r2):
                             for f in a:
                                 if f not in b:
                                     ui.warn('*** inotify: %s +%s\n' % (c, f))
                             for f in b:
                                 if f not in a:
                                     ui.warn('*** inotify: %s -%s\n' % (c, f))
                         result = r2
 
                     if result is not None:
                         return result
             except (OSError, socket.error), err:
                 autostart = ui.configbool('inotify', 'autostart', True)
 
                 if err[0] == errno.ECONNREFUSED:
                     ui.warn(_('(found dead inotify server socket; '
                                    'removing it)\n'))
                     os.unlink(repo.join('inotify.sock'))
                 if err[0] in (errno.ECONNREFUSED, errno.ENOENT) and autostart:
                     ui.debug(_('(starting inotify server)\n'))
                     try:
                         try:
                             server.start(ui, repo)
                         except server.AlreadyStartedException, inst:
                             # another process may have started its own
                             # inotify server while this one was starting.
                             ui.debug(str(inst))
                     except Exception, inst:
                         ui.warn(_('could not start inotify server: '
                                        '%s\n') % inst)
                     else:
                         # server is started, send query again
                         try:
-                            return client.query(ui, repo, files, match,
-                                         ignored, clean, unknown)
+                            return cli.statusquery(files, match, ignored,
+                                                   clean, unknown)
                         except socket.error, err:
                             ui.warn(_('could not talk to new inotify '
                                            'server: %s\n') % err[-1])
                 elif err[0] in (errno.ECONNREFUSED, errno.ENOENT):
                     # silently ignore normal errors if autostart is False
                     ui.debug(_('(inotify server not running)\n'))
                 else:
                     ui.warn(_('failed to contact inotify server: %s\n')
                              % err[-1])
                 ui.traceback()
                 # replace by old status function
                 self.status = super(inotifydirstate, self).status
 
             return super(inotifydirstate, self).status(
                 match, ignored, clean, unknown)
 
     repo.dirstate.__class__ = inotifydirstate
 
 cmdtable = {
     '^inserve':
     (serve,
      [('d', 'daemon', None, _('run server in background')),
       ('', 'daemon-pipefds', '', _('used internally by daemon mode')),
       ('t', 'idle-timeout', '', _('minutes to sit idle before exiting')),
       ('', 'pid-file', '', _('name of file to write process ID to'))],
      _('hg inserve [OPT]...')),
     }

hgext/inotify/client.py

@@ -1,65 +1,94
 # client.py - inotify status client
 #
 # Copyright 2006, 2007, 2008 Bryan O'Sullivan <[e-mail]>
 # Copyright 2007, 2008 Brendan Cully <[e-mail]>
+# Copyright 2009 Nicolas Dumazet <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2, incorporated herein by reference.
 
 from mercurial.i18n import _
 import common
 import os, socket, struct
 
-def query(ui, repo, names, match, ignored, clean, unknown=True):
-    sock = socket.socket(socket.AF_UNIX)
-    sockpath = repo.join('inotify.sock')
-    try:
-        sock.connect(sockpath)
-    except socket.error, err:
-        if err[0] == "AF_UNIX path too long":
-            sockpath = os.readlink(sockpath)
-            sock.connect(sockpath)
-        else:
-            raise
+class client(object):
+    def __init__(self, ui, repo):
+        self.ui = ui
+        self.repo = repo
+        self.sock = socket.socket(socket.AF_UNIX)
+
+    def _connect(self):
+        sockpath = self.repo.join('inotify.sock')
+        try:
+            self.sock.connect(sockpath)
+        except socket.error, err:
+            if err[0] == "AF_UNIX path too long":
+                sockpath = os.readlink(sockpath)
+                self.sock.connect(sockpath)
+            else:
+                raise
 
-    def genquery():
-        for n in names:
-            yield n
-        states = 'almrx!'
-        if ignored:
-            raise ValueError('this is insanity')
-        if clean: states += 'c'
-        if unknown: states += '?'
-        yield states
+    def _send(self, data):
+        """Sends protocol version number, and the data"""
+        self.sock.sendall(chr(common.version) + data)
+
+        self.sock.shutdown(socket.SHUT_WR)
 
-    req = '\0'.join(genquery())
+    def _receive(self):
+        """
+        Read data, check version number, extract headers,
+        and returns a tuple (data descriptor, header)
+        Returns (None, None) on error
+        """
+        cs = common.recvcs(self.sock)
+        version = ord(cs.read(1))
+        if version != common.version:
+            self.ui.warn(_('(inotify: received response from incompatible '
+                      'server version %d)\n') % version)
+            return None, None
 
-    sock.sendall(chr(common.version))
-    sock.sendall(req)
-    sock.shutdown(socket.SHUT_WR)
+        # only one type of request is supported for now
+        type = 'STAT'
+        hdrfmt = common.resphdrfmts[type]
+        hdrsize = common.resphdrsizes[type]
+        try:
+            resphdr = struct.unpack(hdrfmt, cs.read(hdrsize))
+        except struct.error:
+            return None, None
 
-    cs = common.recvcs(sock)
-    version = ord(cs.read(1))
+        return cs, resphdr
+
+    def query(self, req):
+        self._connect()
 
-    if version != common.version:
-        ui.warn(_('(inotify: received response from incompatible server '
-                  'version %d)\n') % version)
-        return None
+        self._send(req)
+
+        return self._receive()
+
+    def statusquery(self, names, match, ignored, clean, unknown=True):
 
-    # only one type of request is supported for now
-    type = 'STAT'
-    hdrfmt = common.resphdrfmts[type]
-    hdrsize = common.resphdrsizes[type]
-    try:
-        resphdr = struct.unpack(hdrfmt, cs.read(hdrsize))
-    except struct.error:
-        return None
+        def genquery():
+            for n in names:
+                yield n
+            states = 'almrx!'
+            if ignored:
+                raise ValueError('this is insanity')
+            if clean: states += 'c'
+            if unknown: states += '?'
+            yield states
+
+        req = '\0'.join(genquery())
 
-    def readnames(nbytes):
-        if nbytes:
-            names = cs.read(nbytes)
-            if names:
-                return filter(match, names.split('\0'))
-        return []
+        cs, resphdr = self.query(req)
+
+        if not cs:
+            return None
 
-    return map(readnames, resphdr)
+        def readnames(nbytes):
+            if nbytes:
+                names = cs.read(nbytes)
+                if names:
+                    return filter(match, names.split('\0'))
+            return []
+
+        return map(readnames, resphdr)