~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2006, 2007, 2008 Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

# TODO: At some point, handle upgrades by just passing the whole request

# across to run on the server.

import bz2

from cStringIO import StringIO

from bzrlib import (

    branch,

    debug,

    errors,

    graph,

    lockdir,

    repository,

    revision,

    symbol_versioning,

)

from bzrlib.branch import BranchReferenceFormat

from bzrlib.bzrdir import BzrDir, RemoteBzrDirFormat

from bzrlib.config import BranchConfig, TreeConfig

from bzrlib.decorators import needs_read_lock, needs_write_lock

from bzrlib.errors import (

    NoSuchRevision,

    SmartProtocolError,

    )

from bzrlib.lockable_files import LockableFiles

from bzrlib.pack import ContainerPushParser

from bzrlib.smart import client, vfs

from bzrlib.symbol_versioning import (

    deprecated_in,

    deprecated_method,

    )

from bzrlib.revision import ensure_null, NULL_REVISION

from bzrlib.trace import mutter, note, warning

# Note: RemoteBzrDirFormat is in bzrdir.py

class RemoteBzrDir(BzrDir):

    """Control directory on a remote server, accessed via bzr:// or similar."""

    def __init__(self, transport, _client=None):

        """Construct a RemoteBzrDir.

        :param _client: Private parameter for testing. Disables probing and the

            use of a real bzrdir.

        """

        BzrDir.__init__(self, transport, RemoteBzrDirFormat())

        # this object holds a delegated bzrdir that uses file-level operations

        # to talk to the other side

        self._real_bzrdir = None

        if _client is None:

            medium = transport.get_smart_medium()

            self._client = client._SmartClient(medium)

        else:

            self._client = _client

            return

        path = self._path_for_remote_call(self._client)

        response = self._client.call('BzrDir.open', path)

        if response not in [('yes',), ('no',)]:

            raise errors.UnexpectedSmartServerResponse(response)

        if response == ('no',):

            raise errors.NotBranchError(path=transport.base)

    def _ensure_real(self):

        """Ensure that there is a _real_bzrdir set.

        Used before calls to self._real_bzrdir.

        """

        if not self._real_bzrdir:

            self._real_bzrdir = BzrDir.open_from_transport(

                self.root_transport, _server_formats=False)

    def create_repository(self, shared=False):

        self._ensure_real()

        self._real_bzrdir.create_repository(shared=shared)

        return self.open_repository()

    def destroy_repository(self):

        """See BzrDir.destroy_repository"""

        self._ensure_real()

        self._real_bzrdir.destroy_repository()

    def create_branch(self):

        self._ensure_real()

        real_branch = self._real_bzrdir.create_branch()

        return RemoteBranch(self, self.find_repository(), real_branch)

    def destroy_branch(self):

        """See BzrDir.destroy_branch"""

        self._ensure_real()

        self._real_bzrdir.destroy_branch()

    def create_workingtree(self, revision_id=None, from_branch=None):

        raise errors.NotLocalUrl(self.transport.base)

    def find_branch_format(self):

        """Find the branch 'format' for this bzrdir.

        This might be a synthetic object for e.g. RemoteBranch and SVN.

        """

        b = self.open_branch()

        return b._format

    def get_branch_reference(self):

        """See BzrDir.get_branch_reference()."""

        path = self._path_for_remote_call(self._client)

        try:

            response = self._client.call('BzrDir.open_branch', path)

        except errors.ErrorFromSmartServer, err:

            if err.error_tuple == ('nobranch',):

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

            raise

        if response[0] == 'ok':

            if response[1] == '':

                # branch at this location.

                return None

            else:

                # a branch reference, use the existing BranchReference logic.

                return response[1]

        else:

            raise errors.UnexpectedSmartServerResponse(response)

    def _get_tree_branch(self):

        """See BzrDir._get_tree_branch()."""

        return None, self.open_branch()

    def open_branch(self, _unsupported=False):

        if _unsupported:

            raise NotImplementedError('unsupported flag support not implemented yet.')

        reference_url = self.get_branch_reference()

        if reference_url is None:

            # branch at this location.

            return RemoteBranch(self, self.find_repository())

        else:

            # a branch reference, use the existing BranchReference logic.

            format = BranchReferenceFormat()

            return format.open(self, _found=True, location=reference_url)

    def open_repository(self):

        path = self._path_for_remote_call(self._client)

        verb = 'BzrDir.find_repositoryV2'

        try:

            try:

                response = self._client.call(verb, path)

            except errors.UnknownSmartMethod:

                verb = 'BzrDir.find_repository'

                response = self._client.call(verb, path)

        except errors.ErrorFromSmartServer, err:

            if err.error_verb == 'norepository':

                raise errors.NoRepositoryPresent(self)

            raise

        if response[0] != 'ok':

            raise errors.UnexpectedSmartServerResponse(response)

        if verb == 'BzrDir.find_repository':

            # servers that don't support the V2 method don't support external

            # references either.

            response = response + ('no', )

        if not (len(response) == 5):

            raise SmartProtocolError('incorrect response length %s' % (response,))

        if response[1] == '':

            format = RemoteRepositoryFormat()

            format.rich_root_data = (response[2] == 'yes')

            format.supports_tree_reference = (response[3] == 'yes')

            # No wire format to check this yet.

            format.supports_external_lookups = (response[4] == 'yes')

            return RemoteRepository(self, format)

        else:

            raise errors.NoRepositoryPresent(self)

    def open_workingtree(self, recommend_upgrade=True):

        self._ensure_real()

        if self._real_bzrdir.has_workingtree():

            raise errors.NotLocalUrl(self.root_transport)

        else:

            raise errors.NoWorkingTree(self.root_transport.base)

    def _path_for_remote_call(self, client):

        """Return the path to be used for this bzrdir in a remote call."""

        return client.remote_path_from_transport(self.root_transport)

    def get_branch_transport(self, branch_format):

        self._ensure_real()

        return self._real_bzrdir.get_branch_transport(branch_format)

    def get_repository_transport(self, repository_format):

        self._ensure_real()

        return self._real_bzrdir.get_repository_transport(repository_format)

    def get_workingtree_transport(self, workingtree_format):

        self._ensure_real()

        return self._real_bzrdir.get_workingtree_transport(workingtree_format)

    def can_convert_format(self):

        """Upgrading of remote bzrdirs is not supported yet."""

        return False

    def needs_format_conversion(self, format=None):

        """Upgrading of remote bzrdirs is not supported yet."""

        return False

    def clone(self, url, revision_id=None, force_new_repo=False):

        self._ensure_real()

        return self._real_bzrdir.clone(url, revision_id=revision_id,

            force_new_repo=force_new_repo)

class RemoteRepositoryFormat(repository.RepositoryFormat):

    """Format for repositories accessed over a _SmartClient.

    Instances of this repository are represented by RemoteRepository

    instances.

    The RemoteRepositoryFormat is parameterized during construction

    to reflect the capabilities of the real, remote format. Specifically

    the attributes rich_root_data and supports_tree_reference are set

    on a per instance basis, and are not set (and should not be) at

    the class level.

    """

    _matchingbzrdir = RemoteBzrDirFormat

    def initialize(self, a_bzrdir, shared=False):

        if not isinstance(a_bzrdir, RemoteBzrDir):

            raise AssertionError('%r is not a RemoteBzrDir' % (a_bzrdir,))

        return a_bzrdir.create_repository(shared=shared)

    def open(self, a_bzrdir):

        if not isinstance(a_bzrdir, RemoteBzrDir):

            raise AssertionError('%r is not a RemoteBzrDir' % (a_bzrdir,))

        return a_bzrdir.open_repository()

    def get_format_description(self):

        return 'bzr remote repository'

    def __eq__(self, other):

        return self.__class__ == other.__class__

    def check_conversion_target(self, target_format):

        if self.rich_root_data and not target_format.rich_root_data:

            raise errors.BadConversionTarget(

                'Does not support rich root data.', target_format)

        if (self.supports_tree_reference and

            not getattr(target_format, 'supports_tree_reference', False)):

            raise errors.BadConversionTarget(

                'Does not support nested trees', target_format)

class RemoteRepository(object):

    """Repository accessed over rpc.

    For the moment most operations are performed using local transport-backed

    Repository objects.

    """

    def __init__(self, remote_bzrdir, format, real_repository=None, _client=None):

        """Create a RemoteRepository instance.

        :param remote_bzrdir: The bzrdir hosting this repository.

        :param format: The RemoteFormat object to use.

        :param real_repository: If not None, a local implementation of the

            repository logic for the repository, usually accessing the data

            via the VFS.

        :param _client: Private testing parameter - override the smart client

            to be used by the repository.

        """

        if real_repository:

            self._real_repository = real_repository

        else:

            self._real_repository = None

        self.bzrdir = remote_bzrdir

        if _client is None:

            self._client = remote_bzrdir._client

        else:

            self._client = _client

        self._format = format

        self._lock_mode = None

        self._lock_token = None

        self._lock_count = 0

        self._leave_lock = False

        # A cache of looked up revision parent data; reset at unlock time.

        self._parents_map = None

        if 'hpss' in debug.debug_flags:

            self._requested_parents = None

        # For tests:

        # These depend on the actual remote format, so force them off for

        # maximum compatibility. XXX: In future these should depend on the

        # remote repository instance, but this is irrelevant until we perform

        # reconcile via an RPC call.

        self._reconcile_does_inventory_gc = False

        self._reconcile_fixes_text_parents = False

        self._reconcile_backsup_inventory = False

        self.base = self.bzrdir.transport.base

    def __str__(self):

        return "%s(%s)" % (self.__class__.__name__, self.base)

    __repr__ = __str__

    def abort_write_group(self):

        """Complete a write group on the decorated repository.

        Smart methods peform operations in a single step so this api

        is not really applicable except as a compatibility thunk

        for older plugins that don't use e.g. the CommitBuilder

        facility.

        """

        self._ensure_real()

        return self._real_repository.abort_write_group()

    def commit_write_group(self):

        """Complete a write group on the decorated repository.

        Smart methods peform operations in a single step so this api

        is not really applicable except as a compatibility thunk

        for older plugins that don't use e.g. the CommitBuilder

        facility.

        """

        self._ensure_real()

        return self._real_repository.commit_write_group()

    def _ensure_real(self):

        """Ensure that there is a _real_repository set.

        Used before calls to self._real_repository.

        """

        if not self._real_repository:

            self.bzrdir._ensure_real()

            #self._real_repository = self.bzrdir._real_bzrdir.open_repository()

            self._set_real_repository(self.bzrdir._real_bzrdir.open_repository())

    def find_text_key_references(self):

        """Find the text key references within the repository.

        :return: a dictionary mapping (file_id, revision_id) tuples to altered file-ids to an iterable of

        revision_ids. Each altered file-ids has the exact revision_ids that

        altered it listed explicitly.

        :return: A dictionary mapping text keys ((fileid, revision_id) tuples)

            to whether they were referred to by the inventory of the

            revision_id that they contain. The inventory texts from all present

            revision ids are assessed to generate this report.

        """

        self._ensure_real()

        return self._real_repository.find_text_key_references()

    def _generate_text_key_index(self):

        """Generate a new text key index for the repository.

        This is an expensive function that will take considerable time to run.

        :return: A dict mapping (file_id, revision_id) tuples to a list of

            parents, also (file_id, revision_id) tuples.

        """

        self._ensure_real()

        return self._real_repository._generate_text_key_index()

    @symbol_versioning.deprecated_method(symbol_versioning.one_four)

    def get_revision_graph(self, revision_id=None):

        """See Repository.get_revision_graph()."""

        return self._get_revision_graph(revision_id)

    def _get_revision_graph(self, revision_id):

        """Private method for using with old (< 1.2) servers to fallback."""

        if revision_id is None:

            revision_id = ''

        elif revision.is_null(revision_id):

            return {}

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

        try:

            response = self._client.call_expecting_body(

                'Repository.get_revision_graph', path, revision_id)

        except errors.ErrorFromSmartServer, err:

            if err.error_verb == 'nosuchrevision':

                raise NoSuchRevision(self, revision_id)

            raise

        response_tuple, response_handler = response

        if response_tuple[0] != 'ok':

            raise errors.UnexpectedSmartServerResponse(response_tuple)

        coded = response_handler.read_body_bytes()

        if coded == '':

            # no revisions in this repository!

            return {}

        lines = coded.split('\n')

        revision_graph = {}

        for line in lines:

            d = tuple(line.split())

            revision_graph[d[0]] = d[1:]

        return revision_graph

    def has_revision(self, revision_id):

        """See Repository.has_revision()."""

        if revision_id == NULL_REVISION:

            # The null revision is always present.

            return True

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

        response = self._client.call(

            'Repository.has_revision', path, revision_id)

        if response[0] not in ('yes', 'no'):

            raise errors.UnexpectedSmartServerResponse(response)

        return response[0] == 'yes'

    def has_revisions(self, revision_ids):

        """See Repository.has_revisions()."""

        result = set()

        for revision_id in revision_ids:

            if self.has_revision(revision_id):

                result.add(revision_id)

        return result

    def has_same_location(self, other):

        return (self.__class__ == other.__class__ and

                self.bzrdir.transport.base == other.bzrdir.transport.base)

    def get_graph(self, other_repository=None):

        """Return the graph for this repository format"""

        parents_provider = self

        if (other_repository is not None and

            other_repository.bzrdir.transport.base !=

            self.bzrdir.transport.base):

            parents_provider = graph._StackedParentsProvider(

                [parents_provider, other_repository._make_parents_provider()])

        return graph.Graph(parents_provider)

    def gather_stats(self, revid=None, committers=None):

        """See Repository.gather_stats()."""

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

        # revid can be None to indicate no revisions, not just NULL_REVISION

        if revid is None or revision.is_null(revid):

            fmt_revid = ''

        else:

            fmt_revid = revid

        if committers is None or not committers:

            fmt_committers = 'no'

        else:

            fmt_committers = 'yes'

        response_tuple, response_handler = self._client.call_expecting_body(

            'Repository.gather_stats', path, fmt_revid, fmt_committers)

        if response_tuple[0] != 'ok':

            raise errors.UnexpectedSmartServerResponse(response_tuple)

        body = response_handler.read_body_bytes()

        result = {}

        for line in body.split('\n'):

            if not line:

                continue

            key, val_text = line.split(':')

            if key in ('revisions', 'size', 'committers'):

                result[key] = int(val_text)

            elif key in ('firstrev', 'latestrev'):

                values = val_text.split(' ')[1:]

                result[key] = (float(values[0]), long(values[1]))

        return result

    def find_branches(self, using=False):

        """See Repository.find_branches()."""

        # should be an API call to the server.

        self._ensure_real()

        return self._real_repository.find_branches(using=using)

    def get_physical_lock_status(self):

        """See Repository.get_physical_lock_status()."""

        # should be an API call to the server.

        self._ensure_real()

        return self._real_repository.get_physical_lock_status()

    def is_in_write_group(self):

        """Return True if there is an open write group.

        write groups are only applicable locally for the smart server..

        """

        if self._real_repository:

            return self._real_repository.is_in_write_group()

    def is_locked(self):

        return self._lock_count >= 1

    def is_shared(self):

        """See Repository.is_shared()."""

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

        response = self._client.call('Repository.is_shared', path)

        if response[0] not in ('yes', 'no'):

            raise SmartProtocolError('unexpected response code %s' % (response,))

        return response[0] == 'yes'

    def is_write_locked(self):

        return self._lock_mode == 'w'

    def lock_read(self):

        # wrong eventually - want a local lock cache context

        if not self._lock_mode:

            self._lock_mode = 'r'

            self._lock_count = 1

            self._parents_map = {}

            if 'hpss' in debug.debug_flags:

                self._requested_parents = set()

            if self._real_repository is not None:

                self._real_repository.lock_read()

        else:

            self._lock_count += 1

    def _remote_lock_write(self, token):

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

        if token is None:

            token = ''

        try:

            response = self._client.call('Repository.lock_write', path, token)

        except errors.ErrorFromSmartServer, err:

            if err.error_verb == 'LockContention':

                raise errors.LockContention('(remote lock)')

            elif err.error_verb == 'UnlockableTransport':

                raise errors.UnlockableTransport(self.bzrdir.root_transport)

            elif err.error_verb == 'LockFailed':

                raise errors.LockFailed(err.error_args[0], err.error_args[1])

            raise

        if response[0] == 'ok':

            ok, token = response

            return token

        else:

            raise errors.UnexpectedSmartServerResponse(response)

    def lock_write(self, token=None):

        if not self._lock_mode:

            self._lock_token = self._remote_lock_write(token)

            # if self._lock_token is None, then this is something like packs or

            # svn where we don't get to lock the repo, or a weave style repository

            # where we cannot lock it over the wire and attempts to do so will

            # fail.

            if self._real_repository is not None:

                self._real_repository.lock_write(token=self._lock_token)

            if token is not None:

                self._leave_lock = True

            else:

                self._leave_lock = False

            self._lock_mode = 'w'

            self._lock_count = 1

            self._parents_map = {}

            if 'hpss' in debug.debug_flags:

                self._requested_parents = set()

        elif self._lock_mode == 'r':

            raise errors.ReadOnlyError(self)

        else:

            self._lock_count += 1

        return self._lock_token or None

    def leave_lock_in_place(self):

        if not self._lock_token:

            raise NotImplementedError(self.leave_lock_in_place)

        self._leave_lock = True

    def dont_leave_lock_in_place(self):

        if not self._lock_token:

            raise NotImplementedError(self.dont_leave_lock_in_place)

        self._leave_lock = False

    def _set_real_repository(self, repository):

        """Set the _real_repository for this repository.

        :param repository: The repository to fallback to for non-hpss

            implemented operations.

        """

        if isinstance(repository, RemoteRepository):

            raise AssertionError()

        self._real_repository = repository

        if self._lock_mode == 'w':

            # if we are already locked, the real repository must be able to

            # acquire the lock with our token.

            self._real_repository.lock_write(self._lock_token)

        elif self._lock_mode == 'r':

            self._real_repository.lock_read()

    def start_write_group(self):

        """Start a write group on the decorated repository.

        Smart methods peform operations in a single step so this api

        is not really applicable except as a compatibility thunk

        for older plugins that don't use e.g. the CommitBuilder

        facility.

        """

        self._ensure_real()

        return self._real_repository.start_write_group()

    def _unlock(self, token):

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

        if not token:

            # with no token the remote repository is not persistently locked.

            return

        try:

            response = self._client.call('Repository.unlock', path, token)

        except errors.ErrorFromSmartServer, err:

            if err.error_verb == 'TokenMismatch':

                raise errors.TokenMismatch(token, '(remote token)')

            raise

        if response == ('ok',):

            return

        else:

            raise errors.UnexpectedSmartServerResponse(response)

    def unlock(self):

        self._lock_count -= 1

        if self._lock_count > 0:

            return

        self._parents_map = None

        if 'hpss' in debug.debug_flags:

            self._requested_parents = None

        old_mode = self._lock_mode

        self._lock_mode = None

        try:

            # The real repository is responsible at present for raising an

            # exception if it's in an unfinished write group.  However, it

            # normally will *not* actually remove the lock from disk - that's

            # done by the server on receiving the Repository.unlock call.

            # This is just to let the _real_repository stay up to date.

            if self._real_repository is not None: