~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2006-2010 Canonical Ltd

    lock,

    repository as _mod_repository,

    static_tuple,

from bzrlib.decorators import needs_read_lock, needs_write_lock, only_raises

    def _call_with_body_bytes(self, method, args, body_bytes, **err_context):

        try:

            return self._client.call_with_body_bytes(method, args, body_bytes)

        except errors.ErrorFromSmartServer, err:

            self._translate_error(err, **err_context)

 

    def __init__(self, transport, format, _client=None, _force_probe=False):

        self._has_working_tree = None

            if not _force_probe:

                return

 

        self._probe_bzrdir()

 

    def __repr__(self):

        return '%s(%r)' % (self.__class__.__name__, self._client)

 

    def _probe_bzrdir(self):

        medium = self._client._medium

        if medium._is_remote_before((2, 1)):

            self._rpc_open(path)

            return

        try:

            self._rpc_open_2_1(path)

            return

        except errors.UnknownSmartMethod:

            medium._remember_remote_is_before((2, 1))

            self._rpc_open(path)

 

    def _rpc_open_2_1(self, path):

        response = self._call('BzrDir.open_2.1', path)

        if response == ('no',):

            raise errors.NotBranchError(path=self.root_transport.base)

        elif response[0] == 'yes':

            if response[1] == 'yes':

                self._has_working_tree = True

            elif response[1] == 'no':

                self._has_working_tree = False

            else:

                raise errors.UnexpectedSmartServerResponse(response)

        else:

            raise errors.UnexpectedSmartServerResponse(response)

 

    def _rpc_open(self, path):

            raise errors.NotBranchError(path=self.root_transport.base)

            if 'hpssvfs' in debug.debug_flags:

                import traceback

                warning('VFS BzrDir access triggered\n%s',

                    ''.join(traceback.format_stack()))

    def create_branch(self, name=None):

        real_branch = self._format.get_branch_format().initialize(self,

            name=name)

            result = RemoteBranch(self, self.find_repository(), real_branch,

                                  name=name)

    def destroy_branch(self, name=None):

        self._real_bzrdir.destroy_branch(name=name)

        candidate_calls = [

            ('BzrDir.open_branchV3', (2, 1)),

            ('BzrDir.open_branchV2', (1, 13)),

            ('BzrDir.open_branch', None),

            ]

        for verb, required_version in candidate_calls:

            if required_version and medium._is_remote_before(required_version):

                continue

                response = self._call(verb, path)

                if required_version is None:

                    raise

                medium._remember_remote_is_before(required_version)

            else:

                break

        if verb == 'BzrDir.open_branch':

            if response[0] != 'ok':

                raise errors.UnexpectedSmartServerResponse(response)

            if response[1] != '':

                return ('ref', response[1])

            else:

                return ('branch', '')

        if response[0] not in ('ref', 'branch'):

        return response

    def open_branch(self, name=None, unsupported=False,

                    ignore_fallbacks=False):

        if unsupported:

            return format.open(self, name=name, _found=True,

                location=response[1], ignore_fallbacks=ignore_fallbacks)

            setup_stacking=not ignore_fallbacks, name=name)

    def has_workingtree(self):

        if self._has_working_tree is None:

            self._ensure_real()

            self._has_working_tree = self._real_bzrdir.has_workingtree()

        return self._has_working_tree

 

        if self.has_workingtree():

    def get_branch_transport(self, branch_format, name=None):

        return self._real_bzrdir.get_branch_transport(branch_format, name=name)

        self._supports_chks = None

    def __repr__(self):

        return "%s(_network_name=%r)" % (self.__class__.__name__,

            self._network_name)

 

    def supports_chks(self):

        if self._supports_chks is None:

            self._ensure_real()

            self._supports_chks = self._custom_format.supports_chks

        return self._supports_chks

 

    @property

        self._ensure_real()

        return 'Remote: ' + self._custom_format.get_format_description()

    def pack_compresses(self):

        self._ensure_real()

        return self._custom_format.pack_compresses

 

    @property

class RemoteRepository(_RpcHelper, lock._RelockDebugMixin,

    bzrdir.ControlComponent):

    @property

    def user_transport(self):

        return self.bzrdir.user_transport

 

    @property

    def control_transport(self):

        # XXX: Normally you shouldn't directly get at the remote repository

        # transport, but I'm not sure it's worth making this method

        # optional -- mbp 2010-04-21

        return self.bzrdir.get_repository_transport(None)

        

    def _get_rev_id_for_revno_vfs(self, revno, known_pair):

        self._ensure_real()

        return self._real_repository.get_rev_id_for_revno(

            revno, known_pair)

 

    def get_rev_id_for_revno(self, revno, known_pair):

        """See Repository.get_rev_id_for_revno."""

        path = self.bzrdir._path_for_remote_call(self._client)

        try:

            if self._client._medium._is_remote_before((1, 17)):

                return self._get_rev_id_for_revno_vfs(revno, known_pair)

            response = self._call(

                'Repository.get_rev_id_for_revno', path, revno, known_pair)

        except errors.UnknownSmartMethod:

            self._client._medium._remember_remote_is_before((1, 17))

            return self._get_rev_id_for_revno_vfs(revno, known_pair)

        if response[0] == 'ok':

            return True, response[1]

        elif response[0] == 'history-incomplete':

            known_pair = response[1:3]

            for fallback in self._fallback_repositories:

                found, result = fallback.get_rev_id_for_revno(revno, known_pair)

                if found:

                    return True, result

                else:

                    known_pair = result

            # Not found in any fallbacks

            return False, known_pair

        else:

            raise errors.UnexpectedSmartServerResponse(response)

 

            if 'hpssvfs' in debug.debug_flags:

    def _has_same_fallbacks(self, other_repo):

        """Returns true if the repositories have the same fallbacks."""

        # XXX: copied from Repository; it should be unified into a base class

        # <https://bugs.edge.launchpad.net/bzr/+bug/401622>

        my_fb = self._fallback_repositories

        other_fb = other_repo._fallback_repositories

        if len(my_fb) != len(other_fb):

            return False

        for f, g in zip(my_fb, other_fb):

            if not f.has_same_location(g):

                return False

        return True

 

        # TODO: Move to RepositoryBase and unify with the regular Repository

        # one; unfortunately the tests rely on slightly different behaviour at

        # present -- mbp 20090710

    @needs_read_lock

    def get_known_graph_ancestry(self, revision_ids):

        """Return the known graph for a set of revision ids and their ancestors.

        """

        st = static_tuple.StaticTuple

        revision_keys = [st(r_id).intern() for r_id in revision_ids]

        known_graph = self.revisions.get_known_graph_ancestry(revision_keys)

        return graph.GraphThunkIdsToKeys(known_graph)

 

    def _warn_if_deprecated(self, branch=None):

        # If we have a real repository, the check will be done there, if we

        # don't the check will be done remotely.

        pass

 

            self._note_lock('r')

            self._note_lock('w')

    @only_raises(errors.LockNotHeld, errors.LockBroken)

            return lock.cant_unlock_not_held(self)

        self._check_fallback_repository(repository)

            fallback_locations = [repo.user_url for repo in

            if repository.user_url not in fallback_locations:

    def _check_fallback_repository(self, repository):

        """Check that this repository can fallback to repository safely.

 

        Raise an error if not.

 

        :param repository: A repository to fallback to.

        """

        return _mod_repository.InterRepository._assert_same_model(

            self, repository)

 

            parents, basis_inv=None, propagate_caches=False):

            delta, new_revision_id, parents, basis_inv=basis_inv,

            propagate_caches=propagate_caches)

    def iter_inventories(self, revision_ids, ordering=None):

        return self._real_repository.iter_inventories(revision_ids, ordering)

        if (self.has_same_location(source)

            and fetch_spec is None

            and self._has_same_fallbacks(source)):

    def _get_inventory_xml(self, revision_id):

        self._ensure_real()

        return self._real_repository._get_inventory_xml(revision_id)

    def check(self, revision_ids=None, callback_refs=None, check_repo=True):

        return self._real_repository.check(revision_ids=revision_ids,

            callback_refs=callback_refs, check_repo=check_repo)

    def pack(self, hint=None, clean_obsolete_packs=False):

        return self._real_repository.pack(hint=hint, clean_obsolete_packs=clean_obsolete_packs)

    def _find_inconsistent_revision_parents(self, revisions_iterator=None):

        return self._real_repository._find_inconsistent_revision_parents(

            revisions_iterator)

        candidate_calls = [('Repository.insert_stream_1.19', (1, 19))]

            candidate_calls.append(('Repository.insert_stream_locked', (1, 14)))

            lock_args = (target._lock_token or '',)

            candidate_calls.append(('Repository.insert_stream', (1, 13)))

            lock_args = ()

        # Probe for the verb to use with an empty stream before sending the

        # real stream to it.  We do this both to avoid the risk of sending a

        # large request that is then rejected, and because we don't want to

        # implement a way to buffer, rewind, or restart the stream.

        found_verb = False

        for verb, required_version in candidate_calls:

            if medium._is_remote_before(required_version):

                continue

            if resume_tokens:

                # We've already done the probing (and set _is_remote_before) on

                # a previous insert.

                found_verb = True

                break

                    (verb, path, '') + lock_args, byte_stream)

            else:

                found_verb = True

                break

        if not found_verb:

            # Have to use VFS.

            return self._insert_real(stream, src_format, resume_tokens)

        self._last_inv_record = None

        self._last_substream = None

        if required_version < (1, 19):

            # Remote side doesn't support inventory deltas.  Wrap the stream to

            # make sure we don't send any.  If the stream contains inventory

            # deltas we'll interrupt the smart insert_stream request and

            # fallback to VFS.

            stream = self._stop_stream_if_inventory_delta(stream)

            (verb, path, resume_tokens) + lock_args, byte_stream)

        if self._last_substream is not None:

            # The stream included an inventory-delta record, but the remote

            # side isn't new enough to support them.  So we need to send the

            # rest of the stream via VFS.

            self.target_repo.refresh_data()

            return self._resume_stream_with_vfs(response, src_format)

    def _resume_stream_with_vfs(self, response, src_format):

        """Resume sending a stream via VFS, first resending the record and

        substream that couldn't be sent via an insert_stream verb.

        """

        if response[0][0] == 'missing-basis':

            tokens, missing_keys = bencode.bdecode_as_tuple(response[0][1])

            # Ignore missing_keys, we haven't finished inserting yet

        else:

            tokens = []

        def resume_substream():

            # Yield the substream that was interrupted.

            for record in self._last_substream:

                yield record

            self._last_substream = None

        def resume_stream():

            # Finish sending the interrupted substream

            yield ('inventory-deltas', resume_substream())

            # Then simply continue sending the rest of the stream.

            for substream_kind, substream in self._last_stream:

                yield substream_kind, substream

        return self._insert_real(resume_stream(), src_format, tokens)

 

    def _stop_stream_if_inventory_delta(self, stream):

        """Normally this just lets the original stream pass-through unchanged.

 

        However if any 'inventory-deltas' substream occurs it will stop

        streaming, and store the interrupted substream and stream in

        self._last_substream and self._last_stream so that the stream can be

        resumed by _resume_stream_with_vfs.

        """

                    

        stream_iter = iter(stream)

        for substream_kind, substream in stream_iter:

            if substream_kind == 'inventory-deltas':

                self._last_substream = substream

                self._last_stream = stream_iter

                return

            else:

                yield substream_kind, substream

            

        sources = []

        seen = set()

        repos = [self.from_repository]

        while repos:

            repo = repos.pop(0)

            if repo in seen:

                continue

            seen.add(repo)

            repos.extend(repo._fallback_repositories)

            sources.append(repo)

        return self.missing_parents_chain(search, sources)

 

    def get_stream_for_missing_keys(self, missing_keys):

        self.from_repository._ensure_real()

        real_repo = self.from_repository._real_repository

        real_source = real_repo._get_source(self.to_format)

        return real_source.get_stream_for_missing_keys(missing_keys)

            repo._ensure_real()

            source = repo._real_repository._get_source(self.to_format)

        search_bytes = repo._serialise_search_result(search)

        args = (path, self.to_format.network_name())

        candidate_verbs = [

            ('Repository.get_stream_1.19', (1, 19)),

            ('Repository.get_stream', (1, 13))]

        found_verb = False

        for verb, version in candidate_verbs:

            if medium._is_remote_before(version):

                continue

            try:

                response = repo._call_with_body_bytes_expecting_body(

                    verb, args, search_bytes)

            except errors.UnknownSmartMethod:

                medium._remember_remote_is_before(version)

            else:

                response_tuple, response_handler = response

                found_verb = True

                break

        if not found_verb:

        self.from_serialiser = self.from_repository._format._serializer

            revision = self.from_serialiser.read_revision_from_string(

                revision_bytes)

        self._ensure_real()

        return 'Remote: ' + self._custom_format.get_format_description()

    def open(self, a_bzrdir, name=None, ignore_fallbacks=False):

        return a_bzrdir.open_branch(name=name, 

            ignore_fallbacks=ignore_fallbacks)

    def _vfs_initialize(self, a_bzrdir, name):

            result = self._custom_format.initialize(a_bzrdir._real_bzrdir,

                name)

            result = self._custom_format.initialize(a_bzrdir, name)

            result = RemoteBranch(a_bzrdir, a_bzrdir.find_repository(), result,

                                  name=name)

    def initialize(self, a_bzrdir, name=None):

            return self._vfs_initialize(a_bzrdir, name=name)

            return self._vfs_initialize(a_bzrdir, name=name)

        if name is not None:

            # XXX JRV20100304: Support creating colocated branches

            raise errors.NoColocatedBranchSupport(self)

            return self._vfs_initialize(a_bzrdir, name=name)

            format=format, setup_stacking=False, name=name)

    def supports_set_append_revisions_only(self):

        self._ensure_real()

        return self._custom_format.supports_set_append_revisions_only()

 

 

class RemoteBranch(branch.Branch, _RpcHelper, lock._RelockDebugMixin):

        _client=None, format=None, setup_stacking=True, name=None):

        :param name: Colocated branch name

        self._clear_cached_state()

        # TODO: deprecate self.base in favor of user_url

        self.base = self.bzrdir.user_url

        self._name = name

        # when we do _ensure_real we may need to pass ignore_fallbacks to the

        # branch.open_branch method.

        self._real_ignore_fallbacks = not setup_stacking

        self._is_stacked = False

        self._is_stacked = True

            self._real_branch = self.bzrdir._real_bzrdir.open_branch(

                ignore_fallbacks=self._real_ignore_fallbacks, name=self._name)

    def set_stacked_on_url(self, url):

        branch.Branch.set_stacked_on_url(self, url)

        if not url:

            self._is_stacked = False

        else:

            self._is_stacked = True

        

    def _vfs_set_tags_bytes(self, bytes):

        self._ensure_real()

        return self._real_branch._set_tags_bytes(bytes)

 

    def _set_tags_bytes(self, bytes):

        medium = self._client._medium

        if medium._is_remote_before((1, 18)):

            self._vfs_set_tags_bytes(bytes)

            return

        try:

            args = (

                self._remote_path(), self._lock_token, self._repo_lock_token)

            response = self._call_with_body_bytes(

                'Branch.set_tags_bytes', args, bytes)

        except errors.UnknownSmartMethod: