client.py

   1 import re, struct
   2 import git
   3 from helpers import *
   4 from subprocess import Popen, PIPE
   5
   6 class ClientError(Exception):
   7     pass
   8
   9 class Client:
  10     def __init__(self, remote, create=False):
  11         self._busy = None
  12         self._indexes_synced = 0
  13         rs = remote.split(':', 1)
  14         if len(rs) == 1:
  15             (host, dir) = ('NONE', remote)
  16             argv = ['bup', 'server']
  17         else:
  18             (host, dir) = rs
  19             argv = ['ssh', host, '--', 'bup', 'server']
  20         (self.host, self.dir) = (host, dir)
  21         self.cachedir = git.repo('index-cache/%s'
  22                                  % re.sub(r'[^@:\w]', '_',
  23                                           "%s:%s" % (host, dir)))
  24         self.p = p = Popen(argv, stdin=PIPE, stdout=PIPE)
  25         self.conn = conn = Conn(p.stdout, p.stdin)
  26         if dir:
  27             dir = re.sub(r'[\r\n]', ' ', dir)
  28             if create:
  29                 conn.write('init-dir %s\n' % dir)
  30             else:
  31                 conn.write('set-dir %s\n' % dir)
  32             conn.check_ok()
  33
  34     def __del__(self):
  35         self.close()
  36
  37     def close(self):
  38         if self.conn and not self._busy:
  39             self.conn.write('quit\n')
  40         if self.p:
  41             self.p.stdin.close()
  42             while self.p.stdout.read(65536):
  43                 pass
  44             self.p.stdout.close()
  45             self.p.wait()
  46             rv = self.p.wait()
  47             if rv:
  48                 raise ClientError('server tunnel returned exit code %d' % rv)
  49         self.conn = None
  50         self.p = None
  51
  52     def check_busy(self):
  53         if self._busy:
  54             raise ClientError('already busy with command %r' % self._busy)
  55
  56     def _not_busy(self):
  57         self._busy = None
  58
  59     def sync_indexes(self):
  60         self.check_busy()
  61         conn = self.conn
  62         conn.write('list-indexes\n')
  63         packdir = git.repo('objects/pack')
  64         mkdirp(self.cachedir)
  65         all = {}
  66         needed = {}
  67         for line in linereader(conn):
  68             if not line:
  69                 break
  70             all[line] = 1
  71             assert(line.find('/') < 0)
  72             if not os.path.exists(os.path.join(self.cachedir, line)):
  73                 needed[line] = 1
  74         conn.check_ok()
  75
  76         for f in os.listdir(self.cachedir):
  77             if f.endswith('.idx') and not f in all:
  78                 log('pruning old index: %r\n' % f)
  79                 os.unlink(os.path.join(self.cachedir, f))
  80
  81         # FIXME this should be pipelined: request multiple indexes at a time, or
  82         # we waste lots of network turnarounds.
  83         for name in needed.keys():
  84             log('requesting %r\n' % name)
  85             conn.write('send-index %s\n' % name)
  86             n = struct.unpack('!I', conn.read(4))[0]
  87             assert(n)
  88             log('   expect %d bytes\n' % n)
  89             fn = os.path.join(self.cachedir, name)
  90             f = open(fn + '.tmp', 'w')
  91             for b in chunkyreader(conn, n):
  92                 f.write(b)
  93             conn.check_ok()
  94             f.close()
  95             os.rename(fn + '.tmp', fn)
  96
  97         self._indexes_synced = 1
  98
  99     def new_packwriter(self):
 100         assert(self._indexes_synced)
 101         self.check_busy()
 102         self._busy = 'receive-objects'
 103         self.conn.write('receive-objects\n')
 104         objcache = git.MultiPackIndex(self.cachedir)
 105         return git.PackWriter_Remote(self.conn, objcache = objcache,
 106                                      onclose = self._not_busy)
 107
 108     def read_ref(self, refname):
 109         self.check_busy()
 110         self.conn.write('read-ref %s\n' % refname)
 111         r = self.conn.readline().strip()
 112         self.conn.check_ok()
 113         if r:
 114             assert(len(r) == 40)   # hexified sha
 115             return r.decode('hex')
 116         else:
 117             return None   # nonexistent ref
 118
 119     def update_ref(self, refname, newval, oldval):
 120         self.check_busy()
 121         self.conn.write('update-ref %s\n%s\n%s\n'
 122                         % (refname, newval.encode('hex'),
 123                            (oldval or '').encode('hex')))
 124         self.conn.check_ok()
 125
 126     def cat(self, id):
 127         self.check_busy()
 128         self._busy = 'cat'
 129         self.conn.write('cat %s\n' % re.sub(r'[\n\r]', '_', id))
 130         while 1:
 131             sz = struct.unpack('!I', self.conn.read(4))[0]
 132             if not sz: break
 133             yield self.conn.read(sz)
 134         self.conn.check_ok()
 135         self._not_busy()