/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/smart/repository.py

  • Committer: Jelmer Vernooij
  • Date: 2011-12-15 11:53:48 UTC
  • mto: This revision was merged to the branch mainline in revision 6375.
  • Revision ID: jelmer@samba.org-20111215115348-murs91ipn8jbw6y0
Add tests for default_email behaviour.

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