Mercurial > hg
view contrib/hgclient.py @ 22814:8110405cf8ae
revset-limit: use boolean testing instead of `len(revs) < 1`
I'm not sure why we wrote it that way. But smartsets have faster/lazier non-zero
testing than length computation.
author | Pierre-Yves David <pierre-yves.david@fb.com> |
---|---|
date | Mon, 06 Oct 2014 11:43:32 -0700 |
parents | cc3d9f776632 |
children | a94594f5d52f |
line wrap: on
line source
# A minimal client for Mercurial's command server import sys, struct, subprocess, cStringIO def connect(path=None): cmdline = ['hg', 'serve', '--cmdserver', 'pipe'] if path: cmdline += ['-R', path] server = subprocess.Popen(cmdline, stdin=subprocess.PIPE, stdout=subprocess.PIPE) return server def writeblock(server, data): server.stdin.write(struct.pack('>I', len(data))) server.stdin.write(data) server.stdin.flush() def readchannel(server): data = server.stdout.read(5) if not data: raise EOFError channel, length = struct.unpack('>cI', data) if channel in 'IL': return channel, length else: return channel, server.stdout.read(length) def sep(text): return text.replace('\\', '/') def runcommand(server, args, output=sys.stdout, error=sys.stderr, input=None, outfilter=lambda x: x): print '*** runcommand', ' '.join(args) sys.stdout.flush() server.stdin.write('runcommand\n') writeblock(server, '\0'.join(args)) if not input: input = cStringIO.StringIO() while True: ch, data = readchannel(server) if ch == 'o': output.write(outfilter(data)) output.flush() elif ch == 'e': error.write(data) error.flush() elif ch == 'I': writeblock(server, input.read(data)) elif ch == 'L': writeblock(server, input.readline(data)) elif ch == 'r': ret, = struct.unpack('>i', data) if ret != 0: print ' [%d]' % ret return ret else: print "unexpected channel %c: %r" % (ch, data) if ch.isupper(): return def check(func, repopath=None): sys.stdout.flush() server = connect(repopath) try: return func(server) finally: server.stdin.close() server.wait()