← Back to branch summary

~bzr/ubuntu/lucid/bzr/beta-ppa

« back to all changes in this revision

Viewing changes to bzrlib/smart/repository.py

  • Committer: Martin Pool
  • Date: 2010-07-02 07:29:40 UTC
  • mfrom: (129.1.7 packaging-karmic)
  • Revision ID: mbp@sourcefrog.net-20100702072940-hpzq5elg8wjve8rh
http://bugs.debian.org/452024
http://bugs.debian.org/464688
http://bugs.debian.org/523595
http://bugs.debian.org/526233
http://bugs.debian.org/537708
http://bugs.debian.org/555336
http://bugs.debian.org/555343
https://launchpad.net/bugs/249452
https://launchpad.net/bugs/385074
* PPA rebuild.
* PPA rebuild for Karmic.
* PPA rebuild for Jaunty.
* PPA rebuild for Hardy.
* From postinst, actually remove the example bash completion scripts.
  (LP: #249452)
* New upstream release.
* New upstream release.
* New upstream release.
* Revert change to Build-depends: Dapper does not have python-central.
  Should be python-support..
* Target ppa..
* Target ppa..
* Target ppa..
* Target ppa..
* New upstream release.
* Switch to dpkg-source 3.0 (quilt) format.
* Bump standards version to 3.8.4.
* Remove embedded copy of python-configobj. Closes: #555336
* Remove embedded copy of python-elementtree. Closes: #555343
* Change section from 'Devel' to 'Vcs'..
* Change section from 'Devel' to 'Vcs'..
* Change section from 'Devel' to 'Vcs'..
* Change section from 'Devel' to 'Vcs'..
* Change section from 'Devel' to 'Vcs'..
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* debian/control: Fix obsolete-relation-form-in-source
  lintian warning. 
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Split out docs into bzr-doc package.
* New upstream release.
* Added John Francesco Ferlito to Uploaders.
* Fix install path to quick-reference guide
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Fix FTBFS due to path changes, again.
* Fix FTBFS due to doc paths changing
* New upstream release.
* Fix FTBFS due to path changes, again.
* Fix FTBFS due to doc paths changing
* New upstream release.
* Fix FTBFS due to path changes, again.
* Fix FTBFS due to doc paths changing
* New upstream release.
* Fix FTBFS due to path changes, again, again.
* Fix FTBFS due to path changes, again.
* Fix FTBFS due to path changes.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Bump standards version to 3.8.3.
* Remove unused patch system.
* New upstream release.
* New upstream release.
* New upstream release.
* Fix copy and paste tab error in .install file
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
 + Fixes compatibility with Python 2.4. Closes: #537708
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream version.
* Bump standards version to 3.8.2.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Add python-pyrex to build-deps to ensure C extensions are always build.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Split documentation into bzr-doc package. ((LP: #385074)
* Multiple packaging changes to make us more linitan clean.
* New upstream release.
* Split documentation into bzr-doc package. ((LP: #385074)
* Multiple packaging changes to make us more linitan clean.
* New upstream release.
* Split documentation into bzr-doc package. ((LP: #385074)
* Multiple packaging changes to make us more linitan clean.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Fix API compatibility version. (Closes: #526233)
* New upstream release.
  + Fixes default format for upgrade command. (Closes: #464688)
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Add missing dependency on zlib development library. (Closes:
  #523595)
* Add zlib build-depends.
* Add zlib build-depends.
* Add zlib build-depends.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Move to section vcs.
* Bump standards version to 3.8.1.
* New upstream release.
* Remove temporary patch for missing .c files from distribution
* New upstream release.
* Remove temporary patch for missing .c files from distribution
* New upstream release.
* Remove temporary patch for missing .c files from distribution
* Add temporary patch for missing .c files from distribution
* Add temporary patch for missing .c files from distribution
* Add temporary patch for missing .c files from distribution
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Recommend ca-certificates. (Closes: #452024)
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Update watch file. bazaar now uses launchpad to host its sources.
* Remove patch for inventory root revision copy, applied upstream.
* New upstream release.
* New upstream release.
* New upstream release
* Force removal of files installed in error to /etc/bash_completion.d/
  (LP: #249452)
* New upstream release.
* New upstream release
* New upstream release.
* Bump standards version.
* Include patch for inventory root revision copy, required for bzr-svn.
* New upstream release.
* Remove unused lintian overrides.
* Correct the package version not to be native.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Final 1.5 release.
* New upstream release.
* New upstream release.
* New upstream release.
* Add myself as a co-maintainer.
* Add a Dm-Upload-Allowed: yes header.
* New upstream bugfix release.
* New upstream release.
* Final 1.3 release.
* New upstream release.
* First release candidate of the upcoming 1.3 release.
* Rebuild to fix the problem caused by a build with a broken python-central.
* New upstream release.
* Rebuild for dapper PPA.
* Apply Lamont's patches to fix build-dependencies on dapper.
  (See: https://bugs.launchpad.net/bzr/+bug/189915)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/repository.py
 
1
# Copyright (C) 2006-2010 Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
 
 
17
"""Server-side repository related request implmentations."""
 
18
 
 
19
import bz2
 
20
import os
 
21
import Queue
 
22
import sys
 
23
import tarfile
 
24
import tempfile
 
25
import threading
 
26
 
 
27
from bzrlib import (
 
28
    bencode,
 
29
    errors,
 
30
    graph,
 
31
    osutils,
 
32
    pack,
 
33
    ui,
 
34
    versionedfile,
 
35
    )
 
36
from bzrlib.bzrdir import BzrDir
 
37
from bzrlib.smart.request import (
 
38
    FailedSmartServerResponse,
 
39
    SmartServerRequest,
 
40
    SuccessfulSmartServerResponse,
 
41
    )
 
42
from bzrlib.repository import _strip_NULL_ghosts, network_format_registry
 
43
from bzrlib import revision as _mod_revision
 
44
from bzrlib.versionedfile import (
 
45
    NetworkRecordStream,
 
46
    record_to_fulltext_bytes,
 
47
    )
 
48
 
 
49
 
 
50
class SmartServerRepositoryRequest(SmartServerRequest):
 
51
    """Common base class for Repository requests."""
 
52
 
 
53
    def do(self, path, *args):
 
54
        """Execute a repository request.
 
55
 
 
56
        All Repository requests take a path to the repository as their first
 
57
        argument.  The repository must be at the exact path given by the
 
58
        client - no searching is done.
 
59
 
 
60
        The actual logic is delegated to self.do_repository_request.
 
61
 
 
62
        :param client_path: The path for the repository as received from the
 
63
            client.
 
64
        :return: A SmartServerResponse from self.do_repository_request().
 
65
        """
 
66
        transport = self.transport_from_client_path(path)
 
67
        bzrdir = BzrDir.open_from_transport(transport)
 
68
        # Save the repository for use with do_body.
 
69
        self._repository = bzrdir.open_repository()
 
70
        return self.do_repository_request(self._repository, *args)
 
71
 
 
72
    def do_repository_request(self, repository, *args):
 
73
        """Override to provide an implementation for a verb."""
 
74
        # No-op for verbs that take bodies (None as a result indicates a body
 
75
        # is expected)
 
76
        return None
 
77
 
 
78
    def recreate_search(self, repository, search_bytes, discard_excess=False):
 
79
        """Recreate a search from its serialised form.
 
80
 
 
81
        :param discard_excess: If True, and the search refers to data we don't
 
82
            have, just silently accept that fact - the verb calling
 
83
            recreate_search trusts that clients will look for missing things
 
84
            they expected and get it from elsewhere.
 
85
        """
 
86
        lines = search_bytes.split('\n')
 
87
        if lines[0] == 'ancestry-of':
 
88
            heads = lines[1:]
 
89
            search_result = graph.PendingAncestryResult(heads, repository)
 
90
            return search_result, None
 
91
        elif lines[0] == 'search':
 
92
            return self.recreate_search_from_recipe(repository, lines[1:],
 
93
                discard_excess=discard_excess)
 
94
        else:
 
95
            return (None, FailedSmartServerResponse(('BadSearch',)))
 
96
 
 
97
    def recreate_search_from_recipe(self, repository, lines,
 
98
        discard_excess=False):
 
99
        """Recreate a specific revision search (vs a from-tip search).
 
100
 
 
101
        :param discard_excess: If True, and the search refers to data we don't
 
102
            have, just silently accept that fact - the verb calling
 
103
            recreate_search trusts that clients will look for missing things
 
104
            they expected and get it from elsewhere.
 
105
        """
 
106
        start_keys = set(lines[0].split(' '))
 
107
        exclude_keys = set(lines[1].split(' '))
 
108
        revision_count = int(lines[2])
 
109
        repository.lock_read()
 
110
        try:
 
111
            search = repository.get_graph()._make_breadth_first_searcher(
 
112
                start_keys)
 
113
            while True:
 
114
                try:
 
115
                    next_revs = search.next()
 
116
                except StopIteration:
 
117
                    break
 
118
                search.stop_searching_any(exclude_keys.intersection(next_revs))
 
119
            search_result = search.get_result()
 
120
            if (not discard_excess and
 
121
                search_result.get_recipe()[3] != revision_count):
 
122
                # we got back a different amount of data than expected, this
 
123
                # gets reported as NoSuchRevision, because less revisions
 
124
                # indicates missing revisions, and more should never happen as
 
125
                # the excludes list considers ghosts and ensures that ghost
 
126
                # filling races are not a problem.
 
127
                return (None, FailedSmartServerResponse(('NoSuchRevision',)))
 
128
            return (search_result, None)
 
129
        finally:
 
130
            repository.unlock()
 
131
 
 
132
 
 
133
class SmartServerRepositoryReadLocked(SmartServerRepositoryRequest):
 
134
    """Calls self.do_readlocked_repository_request."""
 
135
 
 
136
    def do_repository_request(self, repository, *args):
 
137
        """Read lock a repository for do_readlocked_repository_request."""
 
138
        repository.lock_read()
 
139
        try:
 
140
            return self.do_readlocked_repository_request(repository, *args)
 
141
        finally:
 
142
            repository.unlock()
 
143
 
 
144
 
 
145
class SmartServerRepositoryGetParentMap(SmartServerRepositoryRequest):
 
146
    """Bzr 1.2+ - get parent data for revisions during a graph search."""
 
147
 
 
148
    no_extra_results = False
 
149
 
 
150
    def do_repository_request(self, repository, *revision_ids):
 
151
        """Get parent details for some revisions.
 
152
 
 
153
        All the parents for revision_ids are returned. Additionally up to 64KB
 
154
        of additional parent data found by performing a breadth first search
 
155
        from revision_ids is returned. The verb takes a body containing the
 
156
        current search state, see do_body for details.
 
157
 
 
158
        If 'include-missing:' is in revision_ids, ghosts encountered in the
 
159
        graph traversal for getting parent data are included in the result with
 
160
        a prefix of 'missing:'.
 
161
 
 
162
        :param repository: The repository to query in.
 
163
        :param revision_ids: The utf8 encoded revision_id to answer for.
 
164
        """
 
165
        self._revision_ids = revision_ids
 
166
        return None # Signal that we want a body.
 
167
 
 
168
    def do_body(self, body_bytes):
 
169
        """Process the current search state and perform the parent lookup.
 
170
 
 
171
        :return: A smart server response where the body contains an utf8
 
172
            encoded flattened list of the parents of the revisions (the same
 
173
            format as Repository.get_revision_graph) which has been bz2
 
174
            compressed.
 
175
        """
 
176
        repository = self._repository
 
177
        repository.lock_read()
 
178
        try:
 
179
            return self._do_repository_request(body_bytes)
 
180
        finally:
 
181
            repository.unlock()
 
182
 
 
183
    def _do_repository_request(self, body_bytes):
 
184
        repository = self._repository
 
185
        revision_ids = set(self._revision_ids)
 
186
        include_missing = 'include-missing:' in revision_ids
 
187
        if include_missing:
 
188
            revision_ids.remove('include-missing:')
 
189
        body_lines = body_bytes.split('\n')
 
190
        search_result, error = self.recreate_search_from_recipe(
 
191
            repository, body_lines)
 
192
        if error is not None:
 
193
            return error
 
194
        # TODO might be nice to start up the search again; but thats not
 
195
        # written or tested yet.
 
196
        client_seen_revs = set(search_result.get_keys())
 
197
        # Always include the requested ids.
 
198
        client_seen_revs.difference_update(revision_ids)
 
199
        lines = []
 
200
        repo_graph = repository.get_graph()
 
201
        result = {}
 
202
        queried_revs = set()
 
203
        size_so_far = 0
 
204
        next_revs = revision_ids
 
205
        first_loop_done = False
 
206
        while next_revs:
 
207
            queried_revs.update(next_revs)
 
208
            parent_map = repo_graph.get_parent_map(next_revs)
 
209
            current_revs = next_revs
 
210
            next_revs = set()
 
211
            for revision_id in current_revs:
 
212
                missing_rev = False
 
213
                parents = parent_map.get(revision_id)
 
214
                if parents is not None:
 
215
                    # adjust for the wire
 
216
                    if parents == (_mod_revision.NULL_REVISION,):
 
217
                        parents = ()
 
218
                    # prepare the next query
 
219
                    next_revs.update(parents)
 
220
                    encoded_id = revision_id
 
221
                else:
 
222
                    missing_rev = True
 
223
                    encoded_id = "missing:" + revision_id
 
224
                    parents = []
 
225
                if (revision_id not in client_seen_revs and
 
226
                    (not missing_rev or include_missing)):
 
227
                    # Client does not have this revision, give it to it.
 
228
                    # add parents to the result
 
229
                    result[encoded_id] = parents
 
230
                    # Approximate the serialized cost of this revision_id.
 
231
                    size_so_far += 2 + len(encoded_id) + sum(map(len, parents))
 
232
            # get all the directly asked for parents, and then flesh out to
 
233
            # 64K (compressed) or so. We do one level of depth at a time to
 
234
            # stay in sync with the client. The 250000 magic number is
 
235
            # estimated compression ratio taken from bzr.dev itself.
 
236
            if self.no_extra_results or (
 
237
                first_loop_done and size_so_far > 250000):
 
238
                next_revs = set()
 
239
                break
 
240
            # don't query things we've already queried
 
241
            next_revs.difference_update(queried_revs)
 
242
            first_loop_done = True
 
243
 
 
244
        # sorting trivially puts lexographically similar revision ids together.
 
245
        # Compression FTW.
 
246
        for revision, parents in sorted(result.items()):
 
247
            lines.append(' '.join((revision, ) + tuple(parents)))
 
248
 
 
249
        return SuccessfulSmartServerResponse(
 
250
            ('ok', ), bz2.compress('\n'.join(lines)))
 
251
 
 
252
 
 
253
class SmartServerRepositoryGetRevisionGraph(SmartServerRepositoryReadLocked):
 
254
 
 
255
    def do_readlocked_repository_request(self, repository, revision_id):
 
256
        """Return the result of repository.get_revision_graph(revision_id).
 
257
 
 
258
        Deprecated as of bzr 1.4, but supported for older clients.
 
259
 
 
260
        :param repository: The repository to query in.
 
261
        :param revision_id: The utf8 encoded revision_id to get a graph from.
 
262
        :return: A smart server response where the body contains an utf8
 
263
            encoded flattened list of the revision graph.
 
264
        """
 
265
        if not revision_id:
 
266
            revision_id = None
 
267
 
 
268
        lines = []
 
269
        graph = repository.get_graph()
 
270
        if revision_id:
 
271
            search_ids = [revision_id]
 
272
        else:
 
273
            search_ids = repository.all_revision_ids()
 
274
        search = graph._make_breadth_first_searcher(search_ids)
 
275
        transitive_ids = set()
 
276
        map(transitive_ids.update, list(search))
 
277
        parent_map = graph.get_parent_map(transitive_ids)
 
278
        revision_graph = _strip_NULL_ghosts(parent_map)
 
279
        if revision_id and revision_id not in revision_graph:
 
280
            # Note that we return an empty body, rather than omitting the body.
 
281
            # This way the client knows that it can always expect to find a body
 
282
            # in the response for this method, even in the error case.
 
283
            return FailedSmartServerResponse(('nosuchrevision', revision_id), '')
 
284
 
 
285
        for revision, parents in revision_graph.items():
 
286
            lines.append(' '.join((revision, ) + tuple(parents)))
 
287
 
 
288
        return SuccessfulSmartServerResponse(('ok', ), '\n'.join(lines))
 
289
 
 
290
 
 
291
class SmartServerRepositoryGetRevIdForRevno(SmartServerRepositoryReadLocked):
 
292
 
 
293
    def do_readlocked_repository_request(self, repository, revno,
 
294
            known_pair):
 
295
        """Find the revid for a given revno, given a known revno/revid pair.
 
296
        
 
297
        New in 1.17.
 
298
        """
 
299
        try:
 
300
            found_flag, result = repository.get_rev_id_for_revno(revno, known_pair)
 
301
        except errors.RevisionNotPresent, err:
 
302
            if err.revision_id != known_pair[1]:
 
303
                raise AssertionError(
 
304
                    'get_rev_id_for_revno raised RevisionNotPresent for '
 
305
                    'non-initial revision: ' + err.revision_id)
 
306
            return FailedSmartServerResponse(
 
307
                ('nosuchrevision', err.revision_id))
 
308
        if found_flag:
 
309
            return SuccessfulSmartServerResponse(('ok', result))
 
310
        else:
 
311
            earliest_revno, earliest_revid = result
 
312
            return SuccessfulSmartServerResponse(
 
313
                ('history-incomplete', earliest_revno, earliest_revid))
 
314
 
 
315
 
 
316
class SmartServerRequestHasRevision(SmartServerRepositoryRequest):
 
317
 
 
318
    def do_repository_request(self, repository, revision_id):
 
319
        """Return ok if a specific revision is in the repository at path.
 
320
 
 
321
        :param repository: The repository to query in.
 
322
        :param revision_id: The utf8 encoded revision_id to lookup.
 
323
        :return: A smart server response of ('ok', ) if the revision is
 
324
            present.
 
325
        """
 
326
        if repository.has_revision(revision_id):
 
327
            return SuccessfulSmartServerResponse(('yes', ))
 
328
        else:
 
329
            return SuccessfulSmartServerResponse(('no', ))
 
330
 
 
331
 
 
332
class SmartServerRepositoryGatherStats(SmartServerRepositoryRequest):
 
333
 
 
334
    def do_repository_request(self, repository, revid, committers):
 
335
        """Return the result of repository.gather_stats().
 
336
 
 
337
        :param repository: The repository to query in.
 
338
        :param revid: utf8 encoded rev id or an empty string to indicate None
 
339
        :param committers: 'yes' or 'no'.
 
340
 
 
341
        :return: A SmartServerResponse ('ok',), a encoded body looking like
 
342
              committers: 1
 
343
              firstrev: 1234.230 0
 
344
              latestrev: 345.700 3600
 
345
              revisions: 2
 
346
 
 
347
              But containing only fields returned by the gather_stats() call
 
348
        """
 
349
        if revid == '':
 
350
            decoded_revision_id = None
 
351
        else:
 
352
            decoded_revision_id = revid
 
353
        if committers == 'yes':
 
354
            decoded_committers = True
 
355
        else:
 
356
            decoded_committers = None
 
357
        stats = repository.gather_stats(decoded_revision_id, decoded_committers)
 
358
 
 
359
        body = ''
 
360
        if stats.has_key('committers'):
 
361
            body += 'committers: %d\n' % stats['committers']
 
362
        if stats.has_key('firstrev'):
 
363
            body += 'firstrev: %.3f %d\n' % stats['firstrev']
 
364
        if stats.has_key('latestrev'):
 
365
             body += 'latestrev: %.3f %d\n' % stats['latestrev']
 
366
        if stats.has_key('revisions'):
 
367
            body += 'revisions: %d\n' % stats['revisions']
 
368
        if stats.has_key('size'):
 
369
            body += 'size: %d\n' % stats['size']
 
370
 
 
371
        return SuccessfulSmartServerResponse(('ok', ), body)
 
372
 
 
373
 
 
374
class SmartServerRepositoryIsShared(SmartServerRepositoryRequest):
 
375
 
 
376
    def do_repository_request(self, repository):
 
377
        """Return the result of repository.is_shared().
 
378
 
 
379
        :param repository: The repository to query in.
 
380
        :return: A smart server response of ('yes', ) if the repository is
 
381
            shared, and ('no', ) if it is not.
 
382
        """
 
383
        if repository.is_shared():
 
384
            return SuccessfulSmartServerResponse(('yes', ))
 
385
        else:
 
386
            return SuccessfulSmartServerResponse(('no', ))
 
387
 
 
388
 
 
389
class SmartServerRepositoryLockWrite(SmartServerRepositoryRequest):
 
390
 
 
391
    def do_repository_request(self, repository, token=''):
 
392
        # XXX: this probably should not have a token.
 
393
        if token == '':
 
394
            token = None
 
395
        try:
 
396
            token = repository.lock_write(token=token)
 
397
        except errors.LockContention, e:
 
398
            return FailedSmartServerResponse(('LockContention',))
 
399
        except errors.UnlockableTransport:
 
400
            return FailedSmartServerResponse(('UnlockableTransport',))
 
401
        except errors.LockFailed, e:
 
402
            return FailedSmartServerResponse(('LockFailed',
 
403
                str(e.lock), str(e.why)))
 
404
        if token is not None:
 
405
            repository.leave_lock_in_place()
 
406
        repository.unlock()
 
407
        if token is None:
 
408
            token = ''
 
409
        return SuccessfulSmartServerResponse(('ok', token))
 
410
 
 
411
 
 
412
class SmartServerRepositoryGetStream(SmartServerRepositoryRequest):
 
413
 
 
414
    def do_repository_request(self, repository, to_network_name):
 
415
        """Get a stream for inserting into a to_format repository.
 
416
 
 
417
        :param repository: The repository to stream from.
 
418
        :param to_network_name: The network name of the format of the target
 
419
            repository.
 
420
        """
 
421
        self._to_format = network_format_registry.get(to_network_name)
 
422
        if self._should_fake_unknown():
 
423
            return FailedSmartServerResponse(
 
424
                ('UnknownMethod', 'Repository.get_stream'))
 
425
        return None # Signal that we want a body.
 
426
 
 
427
    def _should_fake_unknown(self):
 
428
        """Return True if we should return UnknownMethod to the client.
 
429
        
 
430
        This is a workaround for bugs in pre-1.19 clients that claim to
 
431
        support receiving streams of CHK repositories.  The pre-1.19 client
 
432
        expects inventory records to be serialized in the format defined by
 
433
        to_network_name, but in pre-1.19 (at least) that format definition
 
434
        tries to use the xml5 serializer, which does not correctly handle
 
435
        rich-roots.  After 1.19 the client can also accept inventory-deltas
 
436
        (which avoids this issue), and those clients will use the
 
437
        Repository.get_stream_1.19 verb instead of this one.
 
438
        So: if this repository is CHK, and the to_format doesn't match,
 
439
        we should just fake an UnknownSmartMethod error so that the client
 
440
        will fallback to VFS, rather than sending it a stream we know it
 
441
        cannot handle.
 
442
        """
 
443
        from_format = self._repository._format
 
444
        to_format = self._to_format
 
445
        if not from_format.supports_chks:
 
446
            # Source not CHK: that's ok
 
447
            return False
 
448
        if (to_format.supports_chks and
 
449
            from_format.repository_class is to_format.repository_class and
 
450
            from_format._serializer == to_format._serializer):
 
451
            # Source is CHK, but target matches: that's ok
 
452
            # (e.g. 2a->2a, or CHK2->2a)
 
453
            return False
 
454
        # Source is CHK, and target is not CHK or incompatible CHK.  We can't
 
455
        # generate a compatible stream.
 
456
        return True
 
457
 
 
458
    def do_body(self, body_bytes):
 
459
        repository = self._repository
 
460
        repository.lock_read()
 
461
        try:
 
462
            search_result, error = self.recreate_search(repository, body_bytes,
 
463
                discard_excess=True)
 
464
            if error is not None:
 
465
                repository.unlock()
 
466
                return error
 
467
            source = repository._get_source(self._to_format)
 
468
            stream = source.get_stream(search_result)
 
469
        except Exception:
 
470
            exc_info = sys.exc_info()
 
471
            try:
 
472
                # On non-error, unlocking is done by the body stream handler.
 
473
                repository.unlock()
 
474
            finally:
 
475
                raise exc_info[0], exc_info[1], exc_info[2]
 
476
        return SuccessfulSmartServerResponse(('ok',),
 
477
            body_stream=self.body_stream(stream, repository))
 
478
 
 
479
    def body_stream(self, stream, repository):
 
480
        byte_stream = _stream_to_byte_stream(stream, repository._format)
 
481
        try:
 
482
            for bytes in byte_stream:
 
483
                yield bytes
 
484
        except errors.RevisionNotPresent, e:
 
485
            # This shouldn't be able to happen, but as we don't buffer
 
486
            # everything it can in theory happen.
 
487
            repository.unlock()
 
488
            yield FailedSmartServerResponse(('NoSuchRevision', e.revision_id))
 
489
        else:
 
490
            repository.unlock()
 
491
 
 
492
 
 
493
class SmartServerRepositoryGetStream_1_19(SmartServerRepositoryGetStream):
 
494
 
 
495
    def _should_fake_unknown(self):
 
496
        """Returns False; we don't need to workaround bugs in 1.19+ clients."""
 
497
        return False
 
498
 
 
499
 
 
500
def _stream_to_byte_stream(stream, src_format):
 
501
    """Convert a record stream to a self delimited byte stream."""
 
502
    pack_writer = pack.ContainerSerialiser()
 
503
    yield pack_writer.begin()
 
504
    yield pack_writer.bytes_record(src_format.network_name(), '')
 
505
    for substream_type, substream in stream:
 
506
        for record in substream:
 
507
            if record.storage_kind in ('chunked', 'fulltext'):
 
508
                serialised = record_to_fulltext_bytes(record)
 
509
            elif record.storage_kind == 'inventory-delta':
 
510
                serialised = record_to_inventory_delta_bytes(record)
 
511
            elif record.storage_kind == 'absent':
 
512
                raise ValueError("Absent factory for %s" % (record.key,))
 
513
            else:
 
514
                serialised = record.get_bytes_as(record.storage_kind)
 
515
            if serialised:
 
516
                # Some streams embed the whole stream into the wire
 
517
                # representation of the first record, which means that
 
518
                # later records have no wire representation: we skip them.
 
519
                yield pack_writer.bytes_record(serialised, [(substream_type,)])
 
520
    yield pack_writer.end()
 
521
 
 
522
 
 
523
class _ByteStreamDecoder(object):
 
524
    """Helper for _byte_stream_to_stream.
 
525
 
 
526
    The expected usage of this class is via the function _byte_stream_to_stream
 
527
    which creates a _ByteStreamDecoder, pops off the stream format and then
 
528
    yields the output of record_stream(), the main entry point to
 
529
    _ByteStreamDecoder.
 
530
 
 
531
    Broadly this class has to unwrap two layers of iterators:
 
532
    (type, substream)
 
533
    (substream details)
 
534
 
 
535
    This is complicated by wishing to return type, iterator_for_type, but
 
536
    getting the data for iterator_for_type when we find out type: we can't
 
537
    simply pass a generator down to the NetworkRecordStream parser, instead
 
538
    we have a little local state to seed each NetworkRecordStream instance,
 
539
    and gather the type that we'll be yielding.
 
540
 
 
541
    :ivar byte_stream: The byte stream being decoded.
 
542
    :ivar stream_decoder: A pack parser used to decode the bytestream
 
543
    :ivar current_type: The current type, used to join adjacent records of the
 
544
        same type into a single stream.
 
545
    :ivar first_bytes: The first bytes to give the next NetworkRecordStream.
 
546
    """
 
547
 
 
548
    def __init__(self, byte_stream):
 
549
        """Create a _ByteStreamDecoder."""
 
550
        self.stream_decoder = pack.ContainerPushParser()
 
551
        self.current_type = None
 
552
        self.first_bytes = None
 
553
        self.byte_stream = byte_stream
 
554
 
 
555
    def iter_stream_decoder(self):
 
556
        """Iterate the contents of the pack from stream_decoder."""
 
557
        # dequeue pending items
 
558
        for record in self.stream_decoder.read_pending_records():
 
559
            yield record
 
560
        # Pull bytes of the wire, decode them to records, yield those records.
 
561
        for bytes in self.byte_stream:
 
562
            self.stream_decoder.accept_bytes(bytes)
 
563
            for record in self.stream_decoder.read_pending_records():
 
564
                yield record
 
565
 
 
566
    def iter_substream_bytes(self):
 
567
        if self.first_bytes is not None:
 
568
            yield self.first_bytes
 
569
            # If we run out of pack records, single the outer layer to stop.
 
570
            self.first_bytes = None
 
571
        for record in self.iter_pack_records:
 
572
            record_names, record_bytes = record
 
573
            record_name, = record_names
 
574
            substream_type = record_name[0]
 
575
            if substream_type != self.current_type:
 
576
                # end of a substream, seed the next substream.
 
577
                self.current_type = substream_type
 
578
                self.first_bytes = record_bytes
 
579
                return
 
580
            yield record_bytes
 
581
 
 
582
    def record_stream(self):
 
583
        """Yield substream_type, substream from the byte stream."""
 
584
        self.seed_state()
 
585
        # Make and consume sub generators, one per substream type:
 
586
        while self.first_bytes is not None:
 
587
            substream = NetworkRecordStream(self.iter_substream_bytes())
 
588
            # after substream is fully consumed, self.current_type is set to
 
589
            # the next type, and self.first_bytes is set to the matching bytes.
 
590
            yield self.current_type, substream.read()
 
591
 
 
592
    def seed_state(self):
 
593
        """Prepare the _ByteStreamDecoder to decode from the pack stream."""
 
594
        # Set a single generator we can use to get data from the pack stream.
 
595
        self.iter_pack_records = self.iter_stream_decoder()
 
596
        # Seed the very first subiterator with content; after this each one
 
597
        # seeds the next.
 
598
        list(self.iter_substream_bytes())
 
599
 
 
600
 
 
601
def _byte_stream_to_stream(byte_stream):
 
602
    """Convert a byte stream into a format and a stream.
 
603
 
 
604
    :param byte_stream: A bytes iterator, as output by _stream_to_byte_stream.
 
605
    :return: (RepositoryFormat, stream_generator)
 
606
    """
 
607
    decoder = _ByteStreamDecoder(byte_stream)
 
608
    for bytes in byte_stream:
 
609
        decoder.stream_decoder.accept_bytes(bytes)
 
610
        for record in decoder.stream_decoder.read_pending_records(max=1):
 
611
            record_names, src_format_name = record
 
612
            src_format = network_format_registry.get(src_format_name)
 
613
            return src_format, decoder.record_stream()
 
614
 
 
615
 
 
616
class SmartServerRepositoryUnlock(SmartServerRepositoryRequest):
 
617
 
 
618
    def do_repository_request(self, repository, token):
 
619
        try:
 
620
            repository.lock_write(token=token)
 
621
        except errors.TokenMismatch, e:
 
622
            return FailedSmartServerResponse(('TokenMismatch',))
 
623
        repository.dont_leave_lock_in_place()
 
624
        repository.unlock()
 
625
        return SuccessfulSmartServerResponse(('ok',))
 
626
 
 
627
 
 
628
class SmartServerRepositorySetMakeWorkingTrees(SmartServerRepositoryRequest):
 
629
 
 
630
    def do_repository_request(self, repository, str_bool_new_value):
 
631
        if str_bool_new_value == 'True':
 
632
            new_value = True
 
633
        else:
 
634
            new_value = False
 
635
        repository.set_make_working_trees(new_value)
 
636
        return SuccessfulSmartServerResponse(('ok',))
 
637
 
 
638
 
 
639
class SmartServerRepositoryTarball(SmartServerRepositoryRequest):
 
640
    """Get the raw repository files as a tarball.
 
641
 
 
642
    The returned tarball contains a .bzr control directory which in turn
 
643
    contains a repository.
 
644
 
 
645
    This takes one parameter, compression, which currently must be
 
646
    "", "gz", or "bz2".
 
647
 
 
648
    This is used to implement the Repository.copy_content_into operation.
 
649
    """
 
650
 
 
651
    def do_repository_request(self, repository, compression):
 
652
        tmp_dirname, tmp_repo = self._copy_to_tempdir(repository)
 
653
        try:
 
654
            controldir_name = tmp_dirname + '/.bzr'
 
655
            return self._tarfile_response(controldir_name, compression)
 
656
        finally:
 
657
            osutils.rmtree(tmp_dirname)
 
658
 
 
659
    def _copy_to_tempdir(self, from_repo):
 
660
        tmp_dirname = osutils.mkdtemp(prefix='tmpbzrclone')
 
661
        tmp_bzrdir = from_repo.bzrdir._format.initialize(tmp_dirname)
 
662
        tmp_repo = from_repo._format.initialize(tmp_bzrdir)
 
663
        from_repo.copy_content_into(tmp_repo)
 
664
        return tmp_dirname, tmp_repo
 
665
 
 
666
    def _tarfile_response(self, tmp_dirname, compression):
 
667
        temp = tempfile.NamedTemporaryFile()
 
668
        try:
 
669
            self._tarball_of_dir(tmp_dirname, compression, temp.file)
 
670
            # all finished; write the tempfile out to the network
 
671
            temp.seek(0)
 
672
            return SuccessfulSmartServerResponse(('ok',), temp.read())
 
673
            # FIXME: Don't read the whole thing into memory here; rather stream
 
674
            # it out from the file onto the network. mbp 20070411
 
675
        finally:
 
676
            temp.close()
 
677
 
 
678
    def _tarball_of_dir(self, dirname, compression, ofile):
 
679
        filename = os.path.basename(ofile.name)
 
680
        tarball = tarfile.open(fileobj=ofile, name=filename,
 
681
            mode='w|' + compression)
 
682
        try:
 
683
            # The tarball module only accepts ascii names, and (i guess)
 
684
            # packs them with their 8bit names.  We know all the files
 
685
            # within the repository have ASCII names so the should be safe
 
686
            # to pack in.
 
687
            dirname = dirname.encode(sys.getfilesystemencoding())
 
688
            # python's tarball module includes the whole path by default so
 
689
            # override it
 
690
            if not dirname.endswith('.bzr'):
 
691
                raise ValueError(dirname)
 
692
            tarball.add(dirname, '.bzr') # recursive by default
 
693
        finally:
 
694
            tarball.close()
 
695
 
 
696
 
 
697
class SmartServerRepositoryInsertStreamLocked(SmartServerRepositoryRequest):
 
698
    """Insert a record stream from a RemoteSink into a repository.
 
699
 
 
700
    This gets bytes pushed to it by the network infrastructure and turns that
 
701
    into a bytes iterator using a thread. That is then processed by
 
702
    _byte_stream_to_stream.
 
703
 
 
704
    New in 1.14.
 
705
    """
 
706
 
 
707
    def do_repository_request(self, repository, resume_tokens, lock_token):
 
708
        """StreamSink.insert_stream for a remote repository."""
 
709
        repository.lock_write(token=lock_token)
 
710
        self.do_insert_stream_request(repository, resume_tokens)
 
711
 
 
712
    def do_insert_stream_request(self, repository, resume_tokens):
 
713
        tokens = [token for token in resume_tokens.split(' ') if token]
 
714
        self.tokens = tokens
 
715
        self.repository = repository
 
716
        self.queue = Queue.Queue()
 
717
        self.insert_thread = threading.Thread(target=self._inserter_thread)
 
718
        self.insert_thread.start()
 
719
 
 
720
    def do_chunk(self, body_stream_chunk):
 
721
        self.queue.put(body_stream_chunk)
 
722
 
 
723
    def _inserter_thread(self):
 
724
        try:
 
725
            src_format, stream = _byte_stream_to_stream(
 
726
                self.blocking_byte_stream())
 
727
            self.insert_result = self.repository._get_sink().insert_stream(
 
728
                stream, src_format, self.tokens)
 
729
            self.insert_ok = True
 
730
        except:
 
731
            self.insert_exception = sys.exc_info()
 
732
            self.insert_ok = False
 
733
 
 
734
    def blocking_byte_stream(self):
 
735
        while True:
 
736
            bytes = self.queue.get()
 
737
            if bytes is StopIteration:
 
738
                return
 
739
            else:
 
740
                yield bytes
 
741
 
 
742
    def do_end(self):
 
743
        self.queue.put(StopIteration)
 
744
        if self.insert_thread is not None:
 
745
            self.insert_thread.join()
 
746
        if not self.insert_ok:
 
747
            exc_info = self.insert_exception
 
748
            raise exc_info[0], exc_info[1], exc_info[2]
 
749
        write_group_tokens, missing_keys = self.insert_result
 
750
        if write_group_tokens or missing_keys:
 
751
            # bzip needed? missing keys should typically be a small set.
 
752
            # Should this be a streaming body response ?
 
753
            missing_keys = sorted(missing_keys)
 
754
            bytes = bencode.bencode((write_group_tokens, missing_keys))
 
755
            self.repository.unlock()
 
756
            return SuccessfulSmartServerResponse(('missing-basis', bytes))
 
757
        else:
 
758
            self.repository.unlock()
 
759
            return SuccessfulSmartServerResponse(('ok', ))
 
760
 
 
761
 
 
762
class SmartServerRepositoryInsertStream_1_19(SmartServerRepositoryInsertStreamLocked):
 
763
    """Insert a record stream from a RemoteSink into a repository.
 
764
 
 
765
    Same as SmartServerRepositoryInsertStreamLocked, except:
 
766
     - the lock token argument is optional
 
767
     - servers that implement this verb accept 'inventory-delta' records in the
 
768
       stream.
 
769
 
 
770
    New in 1.19.
 
771
    """
 
772
 
 
773
    def do_repository_request(self, repository, resume_tokens, lock_token=None):
 
774
        """StreamSink.insert_stream for a remote repository."""
 
775
        SmartServerRepositoryInsertStreamLocked.do_repository_request(
 
776
            self, repository, resume_tokens, lock_token)
 
777
 
 
778
 
 
779
class SmartServerRepositoryInsertStream(SmartServerRepositoryInsertStreamLocked):
 
780
    """Insert a record stream from a RemoteSink into an unlocked repository.
 
781
 
 
782
    This is the same as SmartServerRepositoryInsertStreamLocked, except it
 
783
    takes no lock_tokens; i.e. it works with an unlocked (or lock-free, e.g.
 
784
    like pack format) repository.
 
785
 
 
786
    New in 1.13.
 
787
    """
 
788
 
 
789
    def do_repository_request(self, repository, resume_tokens):
 
790
        """StreamSink.insert_stream for a remote repository."""
 
791
        repository.lock_write()
 
792
        self.do_insert_stream_request(repository, resume_tokens)
 
793
 
 
794