# wireproto.py - generic wire protocol support functions

#

# Copyright 2005-2010 Matt Mackall <mpm@selenic.com>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

import urllib, tempfile, os, sys

from i18n import _

from node import bin, hex

import changegroup as changegroupmod, bundle2

import peer, error, encoding, util, store, exchange

class abstractserverproto(object):

    """abstract class that summarizes the protocol API

    Used as reference and documentation.

    """

    def getargs(self, args):

        """return the value for arguments in <args>

        returns a list of values (same order as <args>)"""

        raise NotImplementedError()

    def getfile(self, fp):

        """write the whole content of a file into a file like object

        The file is in the form::

            (<chunk-size>\n<chunk>)+0\n

        chunk size is the ascii version of the int.

        """

        raise NotImplementedError()

    def redirect(self):

        """may setup interception for stdout and stderr

        See also the `restore` method."""

        raise NotImplementedError()

    # If the `redirect` function does install interception, the `restore`

    # function MUST be defined. If interception is not used, this function

    # MUST NOT be defined.

    #

    # left commented here on purpose

    #

    #def restore(self):

    #    """reinstall previous stdout and stderr and return intercepted stdout

    #    """

    #    raise NotImplementedError()

    def groupchunks(self, cg):

        """return 4096 chunks from a changegroup object

        Some protocols may have compressed the contents."""

        raise NotImplementedError()

# abstract batching support

class future(object):

    '''placeholder for a value to be set later'''

    def set(self, value):

        if util.safehasattr(self, 'value'):

            raise error.RepoError("future is already set")

        self.value = value

class batcher(object):

    '''base class for batches of commands submittable in a single request

    All methods invoked on instances of this class are simply queued and

    return a a future for the result. Once you call submit(), all the queued

    calls are performed and the results set in their respective futures.

    '''

    def __init__(self):

        self.calls = []

    def __getattr__(self, name):

        def call(*args, **opts):

            resref = future()

            self.calls.append((name, args, opts, resref,))

            return resref

        return call

    def submit(self):

        pass

class localbatch(batcher):

    '''performs the queued calls directly'''

    def __init__(self, local):

        batcher.__init__(self)

        self.local = local

    def submit(self):

        for name, args, opts, resref in self.calls:

            resref.set(getattr(self.local, name)(*args, **opts))

class remotebatch(batcher):

    '''batches the queued calls; uses as few roundtrips as possible'''

    def __init__(self, remote):

        '''remote must support _submitbatch(encbatch) and

        _submitone(op, encargs)'''

        batcher.__init__(self)

        self.remote = remote

    def submit(self):

        req, rsp = [], []

        for name, args, opts, resref in self.calls:

            mtd = getattr(self.remote, name)

            batchablefn = getattr(mtd, 'batchable', None)

            if batchablefn is not None:

                batchable = batchablefn(mtd.im_self, *args, **opts)

                encargsorres, encresref = batchable.next()

                if encresref:

                    req.append((name, encargsorres,))

                    rsp.append((batchable, encresref, resref,))

                else:

                    resref.set(encargsorres)

            else:

                if req:

                    self._submitreq(req, rsp)

                    req, rsp = [], []

                resref.set(mtd(*args, **opts))

        if req:

            self._submitreq(req, rsp)

    def _submitreq(self, req, rsp):

        encresults = self.remote._submitbatch(req)

        for encres, r in zip(encresults, rsp):

            batchable, encresref, resref = r

            encresref.set(encres)

            resref.set(batchable.next())

def batchable(f):

    '''annotation for batchable methods

    Such methods must implement a coroutine as follows:

    @batchable

    def sample(self, one, two=None):

        # Handle locally computable results first:

        if not one:

            yield "a local result", None

        # Build list of encoded arguments suitable for your wire protocol:

        encargs = [('one', encode(one),), ('two', encode(two),)]

        # Create future for injection of encoded result:

        encresref = future()

        # Return encoded arguments and future:

        yield encargs, encresref

        # Assuming the future to be filled with the result from the batched

        # request now. Decode it:

        yield decode(encresref.value)

    The decorator returns a function which wraps this coroutine as a plain

    method, but adds the original method as an attribute called "batchable",

    which is used by remotebatch to split the call into separate encoding and

    decoding phases.

    '''

    def plain(*args, **opts):

        batchable = f(*args, **opts)

        encargsorres, encresref = batchable.next()

        if not encresref:

            return encargsorres # a local result in this case

        self = args[0]

        encresref.set(self._submitone(f.func_name, encargsorres))

        return batchable.next()

    setattr(plain, 'batchable', f)

    return plain

# list of nodes encoding / decoding

def decodelist(l, sep=' '):

    if l:

        return map(bin, l.split(sep))

    return []

def encodelist(l, sep=' '):

    return sep.join(map(hex, l))

# batched call argument encoding

def escapearg(plain):

    return (plain

            .replace(':', '::')

            .replace(',', ':,')

            .replace(';', ':;')

            .replace('=', ':='))

def unescapearg(escaped):

    return (escaped

            .replace(':=', '=')

            .replace(':;', ';')

            .replace(':,', ',')

            .replace('::', ':'))

# client side

class wirepeer(peer.peerrepository):

    def batch(self):

        return remotebatch(self)

    def _submitbatch(self, req):

        cmds = []

        for op, argsdict in req:

            args = ','.join('%s=%s' % p for p in argsdict.iteritems())

            cmds.append('%s %s' % (op, args))

        rsp = self._call("batch", cmds=';'.join(cmds))

        return rsp.split(';')

    def _submitone(self, op, args):

        return self._call(op, **args)

    @batchable

    def lookup(self, key):

        self.requirecap('lookup', _('look up remote revision'))

        f = future()

        yield {'key': encoding.fromlocal(key)}, f

        d = f.value

        success, data = d[:-1].split(" ", 1)

        if int(success):

            yield bin(data)

        self._abort(error.RepoError(data))

    @batchable

    def heads(self):

        f = future()

        yield {}, f

        d = f.value

        try:

            yield decodelist(d[:-1])

        except ValueError:

            self._abort(error.ResponseError(_("unexpected response:"), d))

    @batchable

    def known(self, nodes):

        f = future()

        yield {'nodes': encodelist(nodes)}, f

        d = f.value

        try:

            yield [bool(int(f)) for f in d]

        except ValueError:

            self._abort(error.ResponseError(_("unexpected response:"), d))

    @batchable

    def branchmap(self):

        f = future()

        yield {}, f

        d = f.value

        try:

            branchmap = {}

            for branchpart in d.splitlines():

                branchname, branchheads = branchpart.split(' ', 1)

                branchname = encoding.tolocal(urllib.unquote(branchname))

                branchheads = decodelist(branchheads)

                branchmap[branchname] = branchheads

            yield branchmap

        except TypeError:

            self._abort(error.ResponseError(_("unexpected response:"), d))

    def branches(self, nodes):

        n = encodelist(nodes)

        d = self._call("branches", nodes=n)

        try:

            br = [tuple(decodelist(b)) for b in d.splitlines()]

            return br

        except ValueError:

            self._abort(error.ResponseError(_("unexpected response:"), d))

    def between(self, pairs):

        batch = 8 # avoid giant requests

        r = []

        for i in xrange(0, len(pairs), batch):

            n = " ".join([encodelist(p, '-') for p in pairs[i:i + batch]])

            d = self._call("between", pairs=n)

            try:

                r.extend(l and decodelist(l) or [] for l in d.splitlines())

            except ValueError:

                self._abort(error.ResponseError(_("unexpected response:"), d))

        return r

    @batchable

    def pushkey(self, namespace, key, old, new):

        if not self.capable('pushkey'):

            yield False, None

        f = future()

        self.ui.debug('preparing pushkey for "%s:%s"\n' % (namespace, key))

        yield {'namespace': encoding.fromlocal(namespace),

               'key': encoding.fromlocal(key),

               'old': encoding.fromlocal(old),

               'new': encoding.fromlocal(new)}, f

        d = f.value

        d, output = d.split('\n', 1)

        try:

            d = bool(int(d))

        except ValueError:

            raise error.ResponseError(

                _('push failed (unexpected response):'), d)

        for l in output.splitlines(True):

            self.ui.status(_('remote: '), l)

        yield d

    @batchable

    def listkeys(self, namespace):

        if not self.capable('pushkey'):

            yield {}, None

        f = future()

        self.ui.debug('preparing listkeys for "%s"\n' % namespace)

        yield {'namespace': encoding.fromlocal(namespace)}, f

        d = f.value

        r = {}

        for l in d.splitlines():

            k, v = l.split('\t')

            r[encoding.tolocal(k)] = encoding.tolocal(v)

        yield r

    def stream_out(self):

        return self._callstream('stream_out')

    def changegroup(self, nodes, kind):

        n = encodelist(nodes)

        f = self._callcompressable("changegroup", roots=n)

        return changegroupmod.unbundle10(f, 'UN')

    def changegroupsubset(self, bases, heads, kind):

        self.requirecap('changegroupsubset', _('look up remote changes'))

        bases = encodelist(bases)

        heads = encodelist(heads)

        f = self._callcompressable("changegroupsubset",

                                   bases=bases, heads=heads)

        return changegroupmod.unbundle10(f, 'UN')

    def getbundle(self, source, heads=None, common=None, bundlecaps=None,

                  **kwargs):

        self.requirecap('getbundle', _('look up remote changes'))

        opts = {}

        if heads is not None:

            opts['heads'] = encodelist(heads)

        if common is not None:

            opts['common'] = encodelist(common)

        if bundlecaps is not None:

            opts['bundlecaps'] = ','.join(bundlecaps)

        opts.update(kwargs)

        f = self._callcompressable("getbundle", **opts)

        if bundlecaps is not None and 'HG2X' in bundlecaps:

            return bundle2.unbundle20(self.ui, f)

        else:

            return changegroupmod.unbundle10(f, 'UN')

    def unbundle(self, cg, heads, source):

        '''Send cg (a readable file-like object representing the

        changegroup to push, typically a chunkbuffer object) to the

        remote server as a bundle.

        When pushing a bundle10 stream, return an integer indicating the

        result of the push (see localrepository.addchangegroup()).

        When pushing a bundle20 stream, return a bundle20 stream.'''

        if heads != ['force'] and self.capable('unbundlehash'):

            heads = encodelist(['hashed',

                                util.sha1(''.join(sorted(heads))).digest()])

        else:

            heads = encodelist(heads)

        if util.safehasattr(cg, 'deltaheader'):

            # this a bundle10, do the old style call sequence

            ret, output = self._callpush("unbundle", cg, heads=heads)

            if ret == "":

                raise error.ResponseError(

                    _('push failed:'), output)

            try:

                ret = int(ret)

            except ValueError:

                raise error.ResponseError(

                    _('push failed (unexpected response):'), ret)

            for l in output.splitlines(True):