← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/repository.py

  • Committer: Vincent Ladeuil
  • Date: 2010-03-10 09:33:04 UTC
  • mto: (5082.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5083.
  • Revision ID: v.ladeuil+lp@free.fr-20100310093304-4245t4tazd4sxoav
Cleanup test from overly cautious checks.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/repository.py
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
 
"""Server-side repository related request implementations."""
18
 
 
19
 
from __future__ import absolute_import
 
17
"""Server-side repository related request implmentations."""
20
18
 
21
19
import bz2
22
20
import os
24
22
import sys
25
23
import tempfile
26
24
import threading
27
 
import zlib
28
25
 
29
26
from bzrlib import (
30
27
    bencode,
31
28
    errors,
32
 
    estimate_compressed_size,
33
 
    inventory as _mod_inventory,
34
 
    inventory_delta,
 
29
    graph,
35
30
    osutils,
36
31
    pack,
37
 
    trace,
38
32
    ui,
39
 
    vf_search,
 
33
    versionedfile,
40
34
    )
41
35
from bzrlib.bzrdir import BzrDir
42
36
from bzrlib.smart.request import (
47
41
from bzrlib.repository import _strip_NULL_ghosts, network_format_registry
48
42
from bzrlib import revision as _mod_revision
49
43
from bzrlib.versionedfile import (
50
 
    ChunkedContentFactory,
51
44
    NetworkRecordStream,
52
45
    record_to_fulltext_bytes,
53
46
    )
89
82
            recreate_search trusts that clients will look for missing things
90
83
            they expected and get it from elsewhere.
91
84
        """
92
 
        if search_bytes == 'everything':
93
 
            return vf_search.EverythingResult(repository), None
94
85
        lines = search_bytes.split('\n')
95
86
        if lines[0] == 'ancestry-of':
96
87
            heads = lines[1:]
97
 
            search_result = vf_search.PendingAncestryResult(heads, repository)
 
88
            search_result = graph.PendingAncestryResult(heads, repository)
98
89
            return search_result, None
99
90
        elif lines[0] == 'search':
100
91
            return self.recreate_search_from_recipe(repository, lines[1:],
124
115
                except StopIteration:
125
116
                    break
126
117
                search.stop_searching_any(exclude_keys.intersection(next_revs))
127
 
            (started_keys, excludes, included_keys) = search.get_state()
128
 
            if (not discard_excess and len(included_keys) != revision_count):
 
118
            search_result = search.get_result()
 
119
            if (not discard_excess and
 
120
                search_result.get_recipe()[3] != revision_count):
129
121
                # we got back a different amount of data than expected, this
130
122
                # gets reported as NoSuchRevision, because less revisions
131
123
                # indicates missing revisions, and more should never happen as
132
124
                # the excludes list considers ghosts and ensures that ghost
133
125
                # filling races are not a problem.
134
126
                return (None, FailedSmartServerResponse(('NoSuchRevision',)))
135
 
            search_result = vf_search.SearchResult(started_keys, excludes,
136
 
                len(included_keys), included_keys)
137
127
            return (search_result, None)
138
128
        finally:
139
129
            repository.unlock()
151
141
            repository.unlock()
152
142
 
153
143
 
154
 
class SmartServerRepositoryBreakLock(SmartServerRepositoryRequest):
155
 
    """Break a repository lock."""
156
 
 
157
 
    def do_repository_request(self, repository):
158
 
        repository.break_lock()
159
 
        return SuccessfulSmartServerResponse(('ok', ))
160
 
 
161
 
 
162
 
_lsprof_count = 0
163
 
 
164
144
class SmartServerRepositoryGetParentMap(SmartServerRepositoryRequest):
165
145
    """Bzr 1.2+ - get parent data for revisions during a graph search."""
166
146
 
199
179
        finally:
200
180
            repository.unlock()
201
181
 
202
 
    def _expand_requested_revs(self, repo_graph, revision_ids, client_seen_revs,
203
 
                               include_missing, max_size=65536):
 
182
    def _do_repository_request(self, body_bytes):
 
183
        repository = self._repository
 
184
        revision_ids = set(self._revision_ids)
 
185
        include_missing = 'include-missing:' in revision_ids
 
186
        if include_missing:
 
187
            revision_ids.remove('include-missing:')
 
188
        body_lines = body_bytes.split('\n')
 
189
        search_result, error = self.recreate_search_from_recipe(
 
190
            repository, body_lines)
 
191
        if error is not None:
 
192
            return error
 
193
        # TODO might be nice to start up the search again; but thats not
 
194
        # written or tested yet.
 
195
        client_seen_revs = set(search_result.get_keys())
 
196
        # Always include the requested ids.
 
197
        client_seen_revs.difference_update(revision_ids)
 
198
        lines = []
 
199
        repo_graph = repository.get_graph()
204
200
        result = {}
205
201
        queried_revs = set()
206
 
        estimator = estimate_compressed_size.ZLibEstimator(max_size)
 
202
        size_so_far = 0
207
203
        next_revs = revision_ids
208
204
        first_loop_done = False
209
205
        while next_revs:
231
227
                    # add parents to the result
232
228
                    result[encoded_id] = parents
233
229
                    # Approximate the serialized cost of this revision_id.
234
 
                    line = '%s %s\n' % (encoded_id, ' '.join(parents))
235
 
                    estimator.add_content(line)
 
230
                    size_so_far += 2 + len(encoded_id) + sum(map(len, parents))
236
231
            # get all the directly asked for parents, and then flesh out to
237
232
            # 64K (compressed) or so. We do one level of depth at a time to
238
233
            # stay in sync with the client. The 250000 magic number is
239
234
            # estimated compression ratio taken from bzr.dev itself.
240
 
            if self.no_extra_results or (first_loop_done and estimator.full()):
241
 
                trace.mutter('size: %d, z_size: %d'
242
 
                             % (estimator._uncompressed_size_added,
243
 
                                estimator._compressed_size_added))
 
235
            if self.no_extra_results or (
 
236
                first_loop_done and size_so_far > 250000):
244
237
                next_revs = set()
245
238
                break
246
239
            # don't query things we've already queried
247
 
            next_revs = next_revs.difference(queried_revs)
 
240
            next_revs.difference_update(queried_revs)
248
241
            first_loop_done = True
249
 
        return result
250
 
 
251
 
    def _do_repository_request(self, body_bytes):
252
 
        repository = self._repository
253
 
        revision_ids = set(self._revision_ids)
254
 
        include_missing = 'include-missing:' in revision_ids
255
 
        if include_missing:
256
 
            revision_ids.remove('include-missing:')
257
 
        body_lines = body_bytes.split('\n')
258
 
        search_result, error = self.recreate_search_from_recipe(
259
 
            repository, body_lines)
260
 
        if error is not None:
261
 
            return error
262
 
        # TODO might be nice to start up the search again; but thats not
263
 
        # written or tested yet.
264
 
        client_seen_revs = set(search_result.get_keys())
265
 
        # Always include the requested ids.
266
 
        client_seen_revs.difference_update(revision_ids)
267
 
 
268
 
        repo_graph = repository.get_graph()
269
 
        result = self._expand_requested_revs(repo_graph, revision_ids,
270
 
                                             client_seen_revs, include_missing)
271
242
 
272
243
        # sorting trivially puts lexographically similar revision ids together.
273
244
        # Compression FTW.
274
 
        lines = []
275
245
        for revision, parents in sorted(result.items()):
276
246
            lines.append(' '.join((revision, ) + tuple(parents)))
277
247
 
342
312
                ('history-incomplete', earliest_revno, earliest_revid))
343
313
 
344
314
 
345
 
class SmartServerRepositoryGetSerializerFormat(SmartServerRepositoryRequest):
346
 
 
347
 
    def do_repository_request(self, repository):
348
 
        """Return the serializer format for this repository.
349
 
 
350
 
        New in 2.5.0.
351
 
 
352
 
        :param repository: The repository to query
353
 
        :return: A smart server response ('ok', FORMAT)
354
 
        """
355
 
        serializer = repository.get_serializer_format()
356
 
        return SuccessfulSmartServerResponse(('ok', serializer))
357
 
 
358
 
 
359
315
class SmartServerRequestHasRevision(SmartServerRepositoryRequest):
360
316
 
361
317
    def do_repository_request(self, repository, revision_id):
363
319
 
364
320
        :param repository: The repository to query in.
365
321
        :param revision_id: The utf8 encoded revision_id to lookup.
366
 
        :return: A smart server response of ('yes', ) if the revision is
367
 
            present. ('no', ) if it is missing.
 
322
        :return: A smart server response of ('ok', ) if the revision is
 
323
            present.
368
324
        """
369
325
        if repository.has_revision(revision_id):
370
326
            return SuccessfulSmartServerResponse(('yes', ))
372
328
            return SuccessfulSmartServerResponse(('no', ))
373
329
 
374
330
 
375
 
class SmartServerRequestHasSignatureForRevisionId(
376
 
        SmartServerRepositoryRequest):
377
 
 
378
 
    def do_repository_request(self, repository, revision_id):
379
 
        """Return ok if a signature is present for a revision.
380
 
 
381
 
        Introduced in bzr 2.5.0.
382
 
 
383
 
        :param repository: The repository to query in.
384
 
        :param revision_id: The utf8 encoded revision_id to lookup.
385
 
        :return: A smart server response of ('yes', ) if a
386
 
            signature for the revision is present,
387
 
            ('no', ) if it is missing.
388
 
        """
389
 
        try:
390
 
            if repository.has_signature_for_revision_id(revision_id):
391
 
                return SuccessfulSmartServerResponse(('yes', ))
392
 
            else:
393
 
                return SuccessfulSmartServerResponse(('no', ))
394
 
        except errors.NoSuchRevision:
395
 
            return FailedSmartServerResponse(
396
 
                ('nosuchrevision', revision_id))
397
 
 
398
 
 
399
331
class SmartServerRepositoryGatherStats(SmartServerRepositoryRequest):
400
332
 
401
333
    def do_repository_request(self, repository, revid, committers):
421
353
            decoded_committers = True
422
354
        else:
423
355
            decoded_committers = None
424
 
        try:
425
 
            stats = repository.gather_stats(decoded_revision_id,
426
 
                decoded_committers)
427
 
        except errors.NoSuchRevision:
428
 
            return FailedSmartServerResponse(('nosuchrevision', revid))
 
356
        stats = repository.gather_stats(decoded_revision_id, decoded_committers)
429
357
 
430
358
        body = ''
431
359
        if stats.has_key('committers'):
442
370
        return SuccessfulSmartServerResponse(('ok', ), body)
443
371
 
444
372
 
445
 
class SmartServerRepositoryGetRevisionSignatureText(
446
 
        SmartServerRepositoryRequest):
447
 
    """Return the signature text of a revision.
448
 
 
449
 
    New in 2.5.
450
 
    """
451
 
 
452
 
    def do_repository_request(self, repository, revision_id):
453
 
        """Return the result of repository.get_signature_text().
454
 
 
455
 
        :param repository: The repository to query in.
456
 
        :return: A smart server response of with the signature text as
457
 
            body.
458
 
        """
459
 
        try:
460
 
            text = repository.get_signature_text(revision_id)
461
 
        except errors.NoSuchRevision, err:
462
 
            return FailedSmartServerResponse(
463
 
                ('nosuchrevision', err.revision))
464
 
        return SuccessfulSmartServerResponse(('ok', ), text)
465
 
 
466
 
 
467
373
class SmartServerRepositoryIsShared(SmartServerRepositoryRequest):
468
374
 
469
375
    def do_repository_request(self, repository):
479
385
            return SuccessfulSmartServerResponse(('no', ))
480
386
 
481
387
 
482
 
class SmartServerRepositoryMakeWorkingTrees(SmartServerRepositoryRequest):
483
 
 
484
 
    def do_repository_request(self, repository):
485
 
        """Return the result of repository.make_working_trees().
486
 
 
487
 
        Introduced in bzr 2.5.0.
488
 
 
489
 
        :param repository: The repository to query in.
490
 
        :return: A smart server response of ('yes', ) if the repository uses
491
 
            working trees, and ('no', ) if it is not.
492
 
        """
493
 
        if repository.make_working_trees():
494
 
            return SuccessfulSmartServerResponse(('yes', ))
495
 
        else:
496
 
            return SuccessfulSmartServerResponse(('no', ))
497
 
 
498
 
 
499
388
class SmartServerRepositoryLockWrite(SmartServerRepositoryRequest):
500
389
 
501
390
    def do_repository_request(self, repository, token=''):
503
392
        if token == '':
504
393
            token = None
505
394
        try:
506
 
            token = repository.lock_write(token=token).repository_token
 
395
            token = repository.lock_write(token=token)
507
396
        except errors.LockContention, e:
508
397
            return FailedSmartServerResponse(('LockContention',))
509
398
        except errors.UnlockableTransport:
524
413
    def do_repository_request(self, repository, to_network_name):
525
414
        """Get a stream for inserting into a to_format repository.
526
415
 
527
 
        The request body is 'search_bytes', a description of the revisions
528
 
        being requested.
529
 
 
530
 
        In 2.3 this verb added support for search_bytes == 'everything'.  Older
531
 
        implementations will respond with a BadSearch error, and clients should
532
 
        catch this and fallback appropriately.
533
 
 
534
416
        :param repository: The repository to stream from.
535
417
        :param to_network_name: The network name of the format of the target
536
418
            repository.
608
490
 
609
491
 
610
492
class SmartServerRepositoryGetStream_1_19(SmartServerRepositoryGetStream):
611
 
    """The same as Repository.get_stream, but will return stream CHK formats to
612
 
    clients.
613
 
 
614
 
    See SmartServerRepositoryGetStream._should_fake_unknown.
615
 
    
616
 
    New in 1.19.
617
 
    """
618
493
 
619
494
    def _should_fake_unknown(self):
620
495
        """Returns False; we don't need to workaround bugs in 1.19+ clients."""
627
502
    yield pack_writer.begin()
628
503
    yield pack_writer.bytes_record(src_format.network_name(), '')
629
504
    for substream_type, substream in stream:
 
505
        if substream_type == 'inventory-deltas':
 
506
            # This doesn't feel like the ideal place to issue this warning;
 
507
            # however we don't want to do it in the Repository that's
 
508
            # generating the stream, because that might be on the server.
 
509
            # Instead we try to observe it as the stream goes by.
 
510
            ui.ui_factory.warn_cross_format_fetch(src_format,
 
511
                '(remote)')
630
512
        for record in substream:
631
513
            if record.storage_kind in ('chunked', 'fulltext'):
632
514
                serialised = record_to_fulltext_bytes(record)
 
515
            elif record.storage_kind == 'inventory-delta':
 
516
                serialised = record_to_inventory_delta_bytes(record)
633
517
            elif record.storage_kind == 'absent':
634
518
                raise ValueError("Absent factory for %s" % (record.key,))
635
519
            else:
667
551
    :ivar first_bytes: The first bytes to give the next NetworkRecordStream.
668
552
    """
669
553
 
670
 
    def __init__(self, byte_stream, record_counter):
 
554
    def __init__(self, byte_stream):
671
555
        """Create a _ByteStreamDecoder."""
672
556
        self.stream_decoder = pack.ContainerPushParser()
673
557
        self.current_type = None
674
558
        self.first_bytes = None
675
559
        self.byte_stream = byte_stream
676
 
        self._record_counter = record_counter
677
 
        self.key_count = 0
678
560
 
679
561
    def iter_stream_decoder(self):
680
562
        """Iterate the contents of the pack from stream_decoder."""
705
587
 
706
588
    def record_stream(self):
707
589
        """Yield substream_type, substream from the byte stream."""
708
 
        def wrap_and_count(pb, rc, substream):
709
 
            """Yield records from stream while showing progress."""
710
 
            counter = 0
711
 
            if rc:
712
 
                if self.current_type != 'revisions' and self.key_count != 0:
713
 
                    # As we know the number of revisions now (in self.key_count)
714
 
                    # we can setup and use record_counter (rc).
715
 
                    if not rc.is_initialized():
716
 
                        rc.setup(self.key_count, self.key_count)
717
 
            for record in substream.read():
718
 
                if rc:
719
 
                    if rc.is_initialized() and counter == rc.STEP:
720
 
                        rc.increment(counter)
721
 
                        pb.update('Estimate', rc.current, rc.max)
722
 
                        counter = 0
723
 
                    if self.current_type == 'revisions':
724
 
                        # Total records is proportional to number of revs
725
 
                        # to fetch. With remote, we used self.key_count to
726
 
                        # track the number of revs. Once we have the revs
727
 
                        # counts in self.key_count, the progress bar changes
728
 
                        # from 'Estimating..' to 'Estimate' above.
729
 
                        self.key_count += 1
730
 
                        if counter == rc.STEP:
731
 
                            pb.update('Estimating..', self.key_count)
732
 
                            counter = 0
733
 
                counter += 1
734
 
                yield record
735
 
 
736
590
        self.seed_state()
737
 
        pb = ui.ui_factory.nested_progress_bar()
738
 
        rc = self._record_counter
739
 
        try:
740
 
            # Make and consume sub generators, one per substream type:
741
 
            while self.first_bytes is not None:
742
 
                substream = NetworkRecordStream(self.iter_substream_bytes())
743
 
                # after substream is fully consumed, self.current_type is set
744
 
                # to the next type, and self.first_bytes is set to the matching
745
 
                # bytes.
746
 
                yield self.current_type, wrap_and_count(pb, rc, substream)
747
 
        finally:
748
 
            if rc:
749
 
                pb.update('Done', rc.max, rc.max)
750
 
            pb.finished()
 
591
        # Make and consume sub generators, one per substream type:
 
592
        while self.first_bytes is not None:
 
593
            substream = NetworkRecordStream(self.iter_substream_bytes())
 
594
            # after substream is fully consumed, self.current_type is set to
 
595
            # the next type, and self.first_bytes is set to the matching bytes.
 
596
            yield self.current_type, substream.read()
751
597
 
752
598
    def seed_state(self):
753
599
        """Prepare the _ByteStreamDecoder to decode from the pack stream."""
758
604
        list(self.iter_substream_bytes())
759
605
 
760
606
 
761
 
def _byte_stream_to_stream(byte_stream, record_counter=None):
 
607
def _byte_stream_to_stream(byte_stream):
762
608
    """Convert a byte stream into a format and a stream.
763
609
 
764
610
    :param byte_stream: A bytes iterator, as output by _stream_to_byte_stream.
765
611
    :return: (RepositoryFormat, stream_generator)
766
612
    """
767
 
    decoder = _ByteStreamDecoder(byte_stream, record_counter)
 
613
    decoder = _ByteStreamDecoder(byte_stream)
768
614
    for bytes in byte_stream:
769
615
        decoder.stream_decoder.accept_bytes(bytes)
770
616
        for record in decoder.stream_decoder.read_pending_records(max=1):
785
631
        return SuccessfulSmartServerResponse(('ok',))
786
632
 
787
633
 
788
 
class SmartServerRepositoryGetPhysicalLockStatus(SmartServerRepositoryRequest):
789
 
    """Get the physical lock status for a repository.
790
 
 
791
 
    New in 2.5.
792
 
    """
793
 
 
794
 
    def do_repository_request(self, repository):
795
 
        if repository.get_physical_lock_status():
796
 
            return SuccessfulSmartServerResponse(('yes', ))
797
 
        else:
798
 
            return SuccessfulSmartServerResponse(('no', ))
799
 
 
800
 
 
801
634
class SmartServerRepositorySetMakeWorkingTrees(SmartServerRepositoryRequest):
802
635
 
803
636
    def do_repository_request(self, repository, str_bool_new_value):
966
799
        self.do_insert_stream_request(repository, resume_tokens)
967
800
 
968
801
 
969
 
class SmartServerRepositoryAddSignatureText(SmartServerRepositoryRequest):
970
 
    """Add a revision signature text.
971
 
 
972
 
    New in 2.5.
973
 
    """
974
 
 
975
 
    def do_repository_request(self, repository, lock_token, revision_id,
976
 
            *write_group_tokens):
977
 
        """Add a revision signature text.
978
 
 
979
 
        :param repository: Repository to operate on
980
 
        :param lock_token: Lock token
981
 
        :param revision_id: Revision for which to add signature
982
 
        :param write_group_tokens: Write group tokens
983
 
        """
984
 
        self._lock_token = lock_token
985
 
        self._revision_id = revision_id
986
 
        self._write_group_tokens = write_group_tokens
987
 
        return None
988
 
 
989
 
    def do_body(self, body_bytes):
990
 
        """Add a signature text.
991
 
 
992
 
        :param body_bytes: GPG signature text
993
 
        :return: SuccessfulSmartServerResponse with arguments 'ok' and
994
 
            the list of new write group tokens.
995
 
        """
996
 
        self._repository.lock_write(token=self._lock_token)
997
 
        try:
998
 
            self._repository.resume_write_group(self._write_group_tokens)
999
 
            try:
1000
 
                self._repository.add_signature_text(self._revision_id,
1001
 
                    body_bytes)
1002
 
            finally:
1003
 
                new_write_group_tokens = self._repository.suspend_write_group()
1004
 
        finally:
1005
 
            self._repository.unlock()
1006
 
        return SuccessfulSmartServerResponse(
1007
 
            ('ok', ) + tuple(new_write_group_tokens))
1008
 
 
1009
 
 
1010
 
class SmartServerRepositoryStartWriteGroup(SmartServerRepositoryRequest):
1011
 
    """Start a write group.
1012
 
 
1013
 
    New in 2.5.
1014
 
    """
1015
 
 
1016
 
    def do_repository_request(self, repository, lock_token):
1017
 
        """Start a write group."""
1018
 
        repository.lock_write(token=lock_token)
1019
 
        try:
1020
 
            repository.start_write_group()
1021
 
            try:
1022
 
                tokens = repository.suspend_write_group()
1023
 
            except errors.UnsuspendableWriteGroup:
1024
 
                return FailedSmartServerResponse(('UnsuspendableWriteGroup',))
1025
 
        finally:
1026
 
            repository.unlock()
1027
 
        return SuccessfulSmartServerResponse(('ok', tokens))
1028
 
 
1029
 
 
1030
 
class SmartServerRepositoryCommitWriteGroup(SmartServerRepositoryRequest):
1031
 
    """Commit a write group.
1032
 
 
1033
 
    New in 2.5.
1034
 
    """
1035
 
 
1036
 
    def do_repository_request(self, repository, lock_token,
1037
 
            write_group_tokens):
1038
 
        """Commit a write group."""
1039
 
        repository.lock_write(token=lock_token)
1040
 
        try:
1041
 
            try:
1042
 
                repository.resume_write_group(write_group_tokens)
1043
 
            except errors.UnresumableWriteGroup, e:
1044
 
                return FailedSmartServerResponse(
1045
 
                    ('UnresumableWriteGroup', e.write_groups, e.reason))
1046
 
            try:
1047
 
                repository.commit_write_group()
1048
 
            except:
1049
 
                write_group_tokens = repository.suspend_write_group()
1050
 
                # FIXME JRV 2011-11-19: What if the write_group_tokens
1051
 
                # have changed?
1052
 
                raise
1053
 
        finally:
1054
 
            repository.unlock()
1055
 
        return SuccessfulSmartServerResponse(('ok', ))
1056
 
 
1057
 
 
1058
 
class SmartServerRepositoryAbortWriteGroup(SmartServerRepositoryRequest):
1059
 
    """Abort a write group.
1060
 
 
1061
 
    New in 2.5.
1062
 
    """
1063
 
 
1064
 
    def do_repository_request(self, repository, lock_token, write_group_tokens):
1065
 
        """Abort a write group."""
1066
 
        repository.lock_write(token=lock_token)
1067
 
        try:
1068
 
            try:
1069
 
                repository.resume_write_group(write_group_tokens)
1070
 
            except errors.UnresumableWriteGroup, e:
1071
 
                return FailedSmartServerResponse(
1072
 
                    ('UnresumableWriteGroup', e.write_groups, e.reason))
1073
 
                repository.abort_write_group()
1074
 
        finally:
1075
 
            repository.unlock()
1076
 
        return SuccessfulSmartServerResponse(('ok', ))
1077
 
 
1078
 
 
1079
 
class SmartServerRepositoryCheckWriteGroup(SmartServerRepositoryRequest):
1080
 
    """Check that a write group is still valid.
1081
 
 
1082
 
    New in 2.5.
1083
 
    """
1084
 
 
1085
 
    def do_repository_request(self, repository, lock_token, write_group_tokens):
1086
 
        """Abort a write group."""
1087
 
        repository.lock_write(token=lock_token)
1088
 
        try:
1089
 
            try:
1090
 
                repository.resume_write_group(write_group_tokens)
1091
 
            except errors.UnresumableWriteGroup, e:
1092
 
                return FailedSmartServerResponse(
1093
 
                    ('UnresumableWriteGroup', e.write_groups, e.reason))
1094
 
            else:
1095
 
                repository.suspend_write_group()
1096
 
        finally:
1097
 
            repository.unlock()
1098
 
        return SuccessfulSmartServerResponse(('ok', ))
1099
 
 
1100
 
 
1101
 
class SmartServerRepositoryAllRevisionIds(SmartServerRepositoryRequest):
1102
 
    """Retrieve all of the revision ids in a repository.
1103
 
 
1104
 
    New in 2.5.
1105
 
    """
1106
 
 
1107
 
    def do_repository_request(self, repository):
1108
 
        revids = repository.all_revision_ids()
1109
 
        return SuccessfulSmartServerResponse(("ok", ), "\n".join(revids))
1110
 
 
1111
 
 
1112
 
class SmartServerRepositoryReconcile(SmartServerRepositoryRequest):
1113
 
    """Reconcile a repository.
1114
 
 
1115
 
    New in 2.5.
1116
 
    """
1117
 
 
1118
 
    def do_repository_request(self, repository, lock_token):
1119
 
        try:
1120
 
            repository.lock_write(token=lock_token)
1121
 
        except errors.TokenLockingNotSupported, e:
1122
 
            return FailedSmartServerResponse(
1123
 
                ('TokenLockingNotSupported', ))
1124
 
        try:
1125
 
            reconciler = repository.reconcile()
1126
 
        finally:
1127
 
            repository.unlock()
1128
 
        body = [
1129
 
            "garbage_inventories: %d\n" % reconciler.garbage_inventories,
1130
 
            "inconsistent_parents: %d\n" % reconciler.inconsistent_parents,
1131
 
            ]
1132
 
        return SuccessfulSmartServerResponse(('ok', ), "".join(body))
1133
 
 
1134
 
 
1135
 
class SmartServerRepositoryPack(SmartServerRepositoryRequest):
1136
 
    """Pack a repository.
1137
 
 
1138
 
    New in 2.5.
1139
 
    """
1140
 
 
1141
 
    def do_repository_request(self, repository, lock_token, clean_obsolete_packs):
1142
 
        self._repository = repository
1143
 
        self._lock_token = lock_token
1144
 
        if clean_obsolete_packs == 'True':
1145
 
            self._clean_obsolete_packs = True
1146
 
        else:
1147
 
            self._clean_obsolete_packs = False
1148
 
        return None
1149
 
 
1150
 
    def do_body(self, body_bytes):
1151
 
        if body_bytes == "":
1152
 
            hint = None
1153
 
        else:
1154
 
            hint = body_bytes.splitlines()
1155
 
        self._repository.lock_write(token=self._lock_token)
1156
 
        try:
1157
 
            self._repository.pack(hint, self._clean_obsolete_packs)
1158
 
        finally:
1159
 
            self._repository.unlock()
1160
 
        return SuccessfulSmartServerResponse(("ok", ), )
1161
 
 
1162
 
 
1163
 
class SmartServerRepositoryIterFilesBytes(SmartServerRepositoryRequest):
1164
 
    """Iterate over the contents of files.
1165
 
 
1166
 
    The client sends a list of desired files to stream, one
1167
 
    per line, and as tuples of file id and revision, separated by
1168
 
    \0.
1169
 
 
1170
 
    The server replies with a stream. Each entry is preceded by a header,
1171
 
    which can either be:
1172
 
 
1173
 
    * "ok\x00IDX\n" where IDX is the index of the entry in the desired files
1174
 
        list sent by the client. This header is followed by the contents of
1175
 
        the file, bzip2-compressed.
1176
 
    * "absent\x00FILEID\x00REVISION\x00IDX" to indicate a text is missing.
1177
 
        The client can then raise an appropriate RevisionNotPresent error
1178
 
        or check its fallback repositories.
1179
 
 
1180
 
    New in 2.5.
1181
 
    """
1182
 
 
1183
 
    def body_stream(self, repository, desired_files):
1184
 
        self._repository.lock_read()
1185
 
        try:
1186
 
            text_keys = {}
1187
 
            for i, key in enumerate(desired_files):
1188
 
                text_keys[key] = i
1189
 
            for record in repository.texts.get_record_stream(text_keys,
1190
 
                    'unordered', True):
1191
 
                identifier = text_keys[record.key]
1192
 
                if record.storage_kind == 'absent':
1193
 
                    yield "absent\0%s\0%s\0%d\n" % (record.key[0],
1194
 
                        record.key[1], identifier)
1195
 
                    # FIXME: Way to abort early?
1196
 
                    continue
1197
 
                yield "ok\0%d\n" % identifier
1198
 
                compressor = zlib.compressobj()
1199
 
                for bytes in record.get_bytes_as('chunked'):
1200
 
                    data = compressor.compress(bytes)
1201
 
                    if data:
1202
 
                        yield data
1203
 
                data = compressor.flush()
1204
 
                if data:
1205
 
                    yield data
1206
 
        finally:
1207
 
            self._repository.unlock()
1208
 
 
1209
 
    def do_body(self, body_bytes):
1210
 
        desired_files = [
1211
 
            tuple(l.split("\0")) for l in body_bytes.splitlines()]
1212
 
        return SuccessfulSmartServerResponse(('ok', ),
1213
 
            body_stream=self.body_stream(self._repository, desired_files))
1214
 
 
1215
 
    def do_repository_request(self, repository):
1216
 
        # Signal that we want a body
1217
 
        return None
1218
 
 
1219
 
 
1220
 
class SmartServerRepositoryIterRevisions(SmartServerRepositoryRequest):
1221
 
    """Stream a list of revisions.
1222
 
 
1223
 
    The client sends a list of newline-separated revision ids in the
1224
 
    body of the request and the server replies with the serializer format,
1225
 
    and a stream of bzip2-compressed revision texts (using the specified
1226
 
    serializer format).
1227
 
 
1228
 
    Any revisions the server does not have are omitted from the stream.
1229
 
 
1230
 
    New in 2.5.
1231
 
    """
1232
 
 
1233
 
    def do_repository_request(self, repository):
1234
 
        self._repository = repository
1235
 
        # Signal there is a body
1236
 
        return None
1237
 
 
1238
 
    def do_body(self, body_bytes):
1239
 
        revision_ids = body_bytes.split("\n")
1240
 
        return SuccessfulSmartServerResponse(
1241
 
            ('ok', self._repository.get_serializer_format()),
1242
 
            body_stream=self.body_stream(self._repository, revision_ids))
1243
 
 
1244
 
    def body_stream(self, repository, revision_ids):
1245
 
        self._repository.lock_read()
1246
 
        try:
1247
 
            for record in repository.revisions.get_record_stream(
1248
 
                [(revid,) for revid in revision_ids], 'unordered', True):
1249
 
                if record.storage_kind == 'absent':
1250
 
                    continue
1251
 
                yield zlib.compress(record.get_bytes_as('fulltext'))
1252
 
        finally:
1253
 
            self._repository.unlock()
1254
 
 
1255
 
 
1256
 
class SmartServerRepositoryGetInventories(SmartServerRepositoryRequest):
1257
 
    """Get the inventory deltas for a set of revision ids.
1258
 
 
1259
 
    This accepts a list of revision ids, and then sends a chain
1260
 
    of deltas for the inventories of those revisions. The first
1261
 
    revision will be empty.
1262
 
 
1263
 
    The server writes back zlibbed serialized inventory deltas,
1264
 
    in the ordering specified. The base for each delta is the
1265
 
    inventory generated by the previous delta.
1266
 
 
1267
 
    New in 2.5.
1268
 
    """
1269
 
 
1270
 
    def _inventory_delta_stream(self, repository, ordering, revids):
1271
 
        prev_inv = _mod_inventory.Inventory(root_id=None,
1272
 
            revision_id=_mod_revision.NULL_REVISION)
1273
 
        serializer = inventory_delta.InventoryDeltaSerializer(
1274
 
            repository.supports_rich_root(),
1275
 
            repository._format.supports_tree_reference)
1276
 
        repository.lock_read()
1277
 
        try:
1278
 
            for inv, revid in repository._iter_inventories(revids, ordering):
1279
 
                if inv is None:
1280
 
                    continue
1281
 
                inv_delta = inv._make_delta(prev_inv)
1282
 
                lines = serializer.delta_to_lines(
1283
 
                    prev_inv.revision_id, inv.revision_id, inv_delta)
1284
 
                yield ChunkedContentFactory(inv.revision_id, None, None, lines)
1285
 
                prev_inv = inv
1286
 
        finally:
1287
 
            repository.unlock()
1288
 
 
1289
 
    def body_stream(self, repository, ordering, revids):
1290
 
        substream = self._inventory_delta_stream(repository,
1291
 
            ordering, revids)
1292
 
        return _stream_to_byte_stream([('inventory-deltas', substream)],
1293
 
            repository._format)
1294
 
 
1295
 
    def do_body(self, body_bytes):
1296
 
        return SuccessfulSmartServerResponse(('ok', ),
1297
 
            body_stream=self.body_stream(self._repository, self._ordering,
1298
 
                body_bytes.splitlines()))
1299
 
 
1300
 
    def do_repository_request(self, repository, ordering):
1301
 
        if ordering == 'unordered':
1302
 
            # inventory deltas for a topologically sorted stream
1303
 
            # are likely to be smaller
1304
 
            ordering = 'topological'
1305
 
        self._ordering = ordering
1306
 
        # Signal that we want a body
1307
 
        return None