# exchange.py - utility to exchange data between repos.

#

# Copyright 2005-2007 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.

from i18n import _

from node import hex, nullid

import errno, urllib

import util, scmutil, changegroup, base85, error

import discovery, phases, obsolete, bookmarks, bundle2, pushkey

def readbundle(ui, fh, fname, vfs=None):

    header = changegroup.readexactly(fh, 4)

    alg = None

    if not fname:

        fname = "stream"

        if not header.startswith('HG') and header.startswith('\0'):

            fh = changegroup.headerlessfixup(fh, header)

            header = "HG10"

            alg = 'UN'

    elif vfs:

        fname = vfs.join(fname)

    magic, version = header[0:2], header[2:4]

    if magic != 'HG':

        raise util.Abort(_('%s: not a Mercurial bundle') % fname)

    if version == '10':

        if alg is None:

            alg = changegroup.readexactly(fh, 2)

        return changegroup.unbundle10(fh, alg)

    elif version == '2X':

        return bundle2.unbundle20(ui, fh, header=magic + version)

    else:

        raise util.Abort(_('%s: unknown bundle version %s') % (fname, version))

class pushoperation(object):

    """A object that represent a single push operation

    It purpose is to carry push related state and very common operation.

    A new should be created at the beginning of each push and discarded

    afterward.

    """

    def __init__(self, repo, remote, force=False, revs=None, newbranch=False):

        # repo we push from

        self.repo = repo

        self.ui = repo.ui

        # repo we push to

        self.remote = remote

        # force option provided

        self.force = force

        # revs to be pushed (None is "all")

        self.revs = revs

        # allow push of new branch

        self.newbranch = newbranch

        # did a local lock get acquired?

        self.locallocked = None

        # step already performed

        # (used to check what steps have been already performed through bundle2)

        self.stepsdone = set()

        # Integer version of the push result

        # - None means nothing to push

        # - 0 means HTTP error

        # - 1 means we pushed and remote head count is unchanged *or*

        #   we have outgoing changesets but refused to push

        # - other values as described by addchangegroup()

        self.ret = None

        # discover.outgoing object (contains common and outgoing data)

        self.outgoing = None

        # all remote heads before the push

        self.remoteheads = None

        # testable as a boolean indicating if any nodes are missing locally.

        self.incoming = None

    @util.propertycache

    def futureheads(self):

        """future remote heads if the changeset push succeeds"""

        return self.outgoing.missingheads

    @util.propertycache

    def fallbackheads(self):

        """future remote heads if the changeset push fails"""

        if self.revs is None:

            # not target to push, all common are relevant

            return self.outgoing.commonheads

        unfi = self.repo.unfiltered()

        # I want cheads = heads(::missingheads and ::commonheads)

        # (missingheads is revs with secret changeset filtered out)

        #

        # This can be expressed as:

        #     cheads = ( (missingheads and ::commonheads)

        #              + (commonheads and ::missingheads))"

        #              )

        #

        # while trying to push we already computed the following:

        #     common = (::commonheads)

        #     missing = ((commonheads::missingheads) - commonheads)

        #

        # We can pick:

        # * missingheads part of common (::commonheads)

        common = set(self.outgoing.common)

        nm = self.repo.changelog.nodemap

        cheads = [node for node in self.revs if nm[node] in common]

        # and

        # * commonheads parents on missing

        revset = unfi.set('%ln and parents(roots(%ln))',

                         self.outgoing.commonheads,

                         self.outgoing.missing)

        cheads.extend(c.node() for c in revset)

        return cheads

    @property

    def commonheads(self):

        """set of all common heads after changeset bundle push"""

        if self.ret:

            return self.futureheads

        else:

            return self.fallbackheads

def push(repo, remote, force=False, revs=None, newbranch=False):

    '''Push outgoing changesets (limited by revs) from a local

    repository to remote. Return an integer:

      - None means nothing to push

      - 0 means HTTP error

      - 1 means we pushed and remote head count is unchanged *or*

        we have outgoing changesets but refused to push

      - other values as described by addchangegroup()

    '''

    pushop = pushoperation(repo, remote, force, revs, newbranch)

    if pushop.remote.local():

        missing = (set(pushop.repo.requirements)

                   - pushop.remote.local().supported)

        if missing:

            msg = _("required features are not"

                    " supported in the destination:"

                    " %s") % (', '.join(sorted(missing)))

            raise util.Abort(msg)

    # there are two ways to push to remote repo:

    #

    # addchangegroup assumes local user can lock remote

    # repo (local filesystem, old ssh servers).

    #

    # unbundle assumes local user cannot lock remote repo (new ssh

    # servers, http servers).

    if not pushop.remote.canpush():

        raise util.Abort(_("destination does not support push"))

    # get local lock as we might write phase data

    locallock = None

    try:

        locallock = pushop.repo.lock()

        pushop.locallocked = True

    except IOError, err:

        pushop.locallocked = False

        if err.errno != errno.EACCES:

            raise

        # source repo cannot be locked.

        # We do not abort the push, but just disable the local phase

        # synchronisation.

        msg = 'cannot lock source repository: %s\n' % err

        pushop.ui.debug(msg)

    try:

        pushop.repo.checkpush(pushop)

        lock = None

        unbundle = pushop.remote.capable('unbundle')

        if not unbundle:

            lock = pushop.remote.lock()

        try:

            _pushdiscovery(pushop)

            if (pushop.repo.ui.configbool('experimental', 'bundle2-exp',

                                          False)

                and pushop.remote.capable('bundle2-exp')):

                _pushbundle2(pushop)

            _pushchangeset(pushop)

            _pushsyncphase(pushop)

            _pushobsolete(pushop)

        finally:

            if lock is not None:

                lock.release()

    finally:

        if locallock is not None:

            locallock.release()

    _pushbookmark(pushop)

    return pushop.ret

def _pushdiscovery(pushop):

    unfi = pushop.repo.unfiltered()

    fci = discovery.findcommonincoming

    commoninc = fci(unfi, pushop.remote, force=pushop.force)

    common, inc, remoteheads = commoninc

    fco = discovery.findcommonoutgoing

    outgoing = fco(unfi, pushop.remote, onlyheads=pushop.revs,

                   commoninc=commoninc, force=pushop.force)

    pushop.outgoing = outgoing

    pushop.remoteheads = remoteheads

    pushop.incoming = inc

def _pushcheckoutgoing(pushop):

    outgoing = pushop.outgoing

    unfi = pushop.repo.unfiltered()

    if not outgoing.missing:

        # nothing to push

        scmutil.nochangesfound(unfi.ui, unfi, outgoing.excluded)

        return False

    # something to push

    if not pushop.force:

        # if repo.obsstore == False --> no obsolete

        # then, save the iteration

        if unfi.obsstore:

            # this message are here for 80 char limit reason

            mso = _("push includes obsolete changeset: %s!")

            mst = "push includes %s changeset: %s!"

            # plain versions for i18n tool to detect them

            _("push includes unstable changeset: %s!")

            _("push includes bumped changeset: %s!")

            _("push includes divergent changeset: %s!")

            # If we are to push if there is at least one

            # obsolete or unstable changeset in missing, at

            # least one of the missinghead will be obsolete or

            # unstable. So checking heads only is ok

            for node in outgoing.missingheads:

                ctx = unfi[node]

                if ctx.obsolete():

                    raise util.Abort(mso % ctx)

                elif ctx.troubled():

                    raise util.Abort(_(mst)

                                     % (ctx.troubles()[0],

                                        ctx))

        newbm = pushop.ui.configlist('bookmarks', 'pushing')

        discovery.checkheads(unfi, pushop.remote, outgoing,

                             pushop.remoteheads,

                             pushop.newbranch,

                             bool(pushop.incoming),

                             newbm)

    return True

# List of names of steps to perform for an outgoing bundle2, order matters.

b2partsgenorder = []

# Mapping between step name and function

#

# This exists to help extensions wrap steps if necessary

b2partsgenmapping = {}

def b2partsgenerator(stepname):

    """decorator for function generating bundle2 part

    The function is added to the step -> function mapping and appended to the

    list of steps.  Beware that decorated functions will be added in order

    (this may matter).

    You can only use this decorator for new steps, if you want to wrap a step

    from an extension, attack the b2partsgenmapping dictionary directly."""

    def dec(func):

        assert stepname not in b2partsgenmapping

        b2partsgenmapping[stepname] = func

        b2partsgenorder.append(stepname)

        return func

    return dec

@b2partsgenerator('changeset')

def _pushb2ctx(pushop, bundler):

    """handle changegroup push through bundle2

    addchangegroup result is stored in the ``pushop.ret`` attribute.

    """

    if 'changesets' in pushop.stepsdone:

        return

    pushop.stepsdone.add('changesets')

    # Send known heads to the server for race detection.

    pushop.stepsdone.add('changesets')

    if not _pushcheckoutgoing(pushop):

        return

    pushop.repo.prepushoutgoinghooks(pushop.repo,

                                     pushop.remote,

                                     pushop.outgoing)

    if not pushop.force:

        bundler.newpart('B2X:CHECK:HEADS', data=iter(pushop.remoteheads))

    cg = changegroup.getlocalbundle(pushop.repo, 'push', pushop.outgoing)

    cgpart = bundler.newpart('B2X:CHANGEGROUP', data=cg.getchunks())

    def handlereply(op):

        """extract addchangroup returns from server reply"""

        cgreplies = op.records.getreplies(cgpart.id)

        assert len(cgreplies['changegroup']) == 1

        pushop.ret = cgreplies['changegroup'][0]['return']

    return handlereply

def _pushbundle2(pushop):

    """push data to the remote using bundle2

    The only currently supported type of data is changegroup but this will

    evolve in the future."""

    bundler = bundle2.bundle20(pushop.ui, bundle2.bundle2caps(pushop.remote))

    # create reply capability

    capsblob = bundle2.encodecaps(pushop.repo.bundle2caps)

    bundler.newpart('b2x:replycaps', data=capsblob)

    replyhandlers = []

    for partgenname in b2partsgenorder:

        partgen = b2partsgenmapping[partgenname]

        ret = partgen(pushop, bundler)

        if callable(ret):

            replyhandlers.append(ret)

    # do not push if nothing to push

    if bundler.nbparts <= 1:

        return

    stream = util.chunkbuffer(bundler.getchunks())

    try:

        reply = pushop.remote.unbundle(stream, ['force'], 'push')

    except error.BundleValueError, exc:

        raise util.Abort('missing support for %s' % exc)

    try:

        op = bundle2.processbundle(pushop.repo, reply)

    except error.BundleValueError, exc:

        raise util.Abort('missing support for %s' % exc)

    for rephand in replyhandlers:

        rephand(op)

def _pushchangeset(pushop):

    """Make the actual push of changeset bundle to remote repo"""

    if 'changesets' in pushop.stepsdone:

        return

    pushop.stepsdone.add('changesets')

    if not _pushcheckoutgoing(pushop):

        return

    pushop.repo.prepushoutgoinghooks(pushop.repo,

                                     pushop.remote,

                                     pushop.outgoing)

    outgoing = pushop.outgoing

    unbundle = pushop.remote.capable('unbundle')

    # TODO: get bundlecaps from remote

    bundlecaps = None

    # create a changegroup from local

    if pushop.revs is None and not (outgoing.excluded

                            or pushop.repo.changelog.filteredrevs):

        # push everything,

        # use the fast path, no race possible on push

        bundler = changegroup.bundle10(pushop.repo, bundlecaps)

        cg = changegroup.getsubset(pushop.repo,

                                   outgoing,

                                   bundler,

                                   'push',

                                   fastpath=True)

    else:

        cg = changegroup.getlocalbundle(pushop.repo, 'push', outgoing,

                                        bundlecaps)