# 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

        # phases changes that must be pushed along side the changesets

        self.outdatedphases = None

        # phases changes that must be pushed if changeset push fails

        self.fallbackoutdatedphases = None

        # outgoing obsmarkers

        self.outobsmarkers = set()

    @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)

            _pushbookmark(pushop)

        finally:

            if lock is not None:

                lock.release()

    finally:

        if locallock is not None:

            locallock.release()

    return pushop.ret

# list of steps to perform discovery before push

pushdiscoveryorder = []

# Mapping between step name and function

#

# This exists to help extensions wrap steps if necessary

pushdiscoverymapping = {}

def pushdiscovery(stepname):

    """decorator for function performing discovery before push

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

    list of steps.  Beware that decorated function will be added in order (this

    may matter).

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

    from an extension, change the pushdiscovery dictionary directly."""

    def dec(func):

        assert stepname not in pushdiscoverymapping

        pushdiscoverymapping[stepname] = func

        pushdiscoveryorder.append(stepname)

        return func

    return dec

def _pushdiscovery(pushop):

    """Run all discovery steps"""

    for stepname in pushdiscoveryorder:

        step = pushdiscoverymapping[stepname]

        step(pushop)

@pushdiscovery('changeset')

def _pushdiscoverychangeset(pushop):

    """discover the changeset that need to be pushed"""

    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

@pushdiscovery('phase')

def _pushdiscoveryphase(pushop):

    """discover the phase that needs to be pushed

    (computed for both success and failure case for changesets push)"""

    outgoing = pushop.outgoing

    unfi = pushop.repo.unfiltered()

    remotephases = pushop.remote.listkeys('phases')

    publishing = remotephases.get('publishing', False)

    ana = phases.analyzeremotephases(pushop.repo,

                                     pushop.fallbackheads,

                                     remotephases)

    pheads, droots = ana

    extracond = ''

    if not publishing:

        extracond = ' and public()'

    revset = 'heads((%%ln::%%ln) %s)' % extracond

    # Get the list of all revs draft on remote by public here.

    # XXX Beware that revset break if droots is not strictly

    # XXX root we may want to ensure it is but it is costly

    fallback = list(unfi.set(revset, droots, pushop.fallbackheads))

    if not outgoing.missing:

        future = fallback

    else:

        # adds changeset we are going to push as draft

        #

        # should not be necessary for pushblishing server, but because of an

        # issue fixed in xxxxx we have to do it anyway.

        fdroots = list(unfi.set('roots(%ln  + %ln::)',

                       outgoing.missing, droots))

        fdroots = [f.node() for f in fdroots]

        future = list(unfi.set(revset, fdroots, pushop.futureheads))

    pushop.outdatedphases = future

    pushop.fallbackoutdatedphases = fallback

@pushdiscovery('obsmarker')

def _pushdiscoveryobsmarkers(pushop):

    pushop.outobsmarkers = pushop.repo.obsstore

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

@b2partsgenerator('phase')

def _pushb2phases(pushop, bundler):

    """handle phase push through bundle2"""

    if 'phases' in pushop.stepsdone:

        return

    b2caps = bundle2.bundle2caps(pushop.remote)

    if not 'b2x:pushkey' in b2caps:

        return

    pushop.stepsdone.add('phases')

    part2node = []

    enc = pushkey.encode

    for newremotehead in pushop.outdatedphases:

        part = bundler.newpart('b2x:pushkey')

        part.addparam('namespace', enc('phases'))

        part.addparam('key', enc(newremotehead.hex()))

        part.addparam('old', enc(str(phases.draft)))

        part.addparam('new', enc(str(phases.public)))