← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/repository.py

(vila) Make all transport put_bytes() raises TypeError when given unicode
 strings rather than bytes (Vincent Ladeuil)

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
 
 
19
from __future__ import absolute_import
18
20
 
19
21
import bz2
20
22
import os
22
24
import sys
23
25
import tempfile
24
26
import threading
 
27
import zlib
25
28
 
26
29
from bzrlib import (
27
30
    bencode,
28
31
    errors,
29
 
    graph,
 
32
    estimate_compressed_size,
 
33
    inventory as _mod_inventory,
 
34
    inventory_delta,
30
35
    osutils,
31
36
    pack,
 
37
    trace,
32
38
    ui,
33
 
    versionedfile,
 
39
    vf_search,
34
40
    )
35
41
from bzrlib.bzrdir import BzrDir
36
42
from bzrlib.smart.request import (
41
47
from bzrlib.repository import _strip_NULL_ghosts, network_format_registry
42
48
from bzrlib import revision as _mod_revision
43
49
from bzrlib.versionedfile import (
 
50
    ChunkedContentFactory,
44
51
    NetworkRecordStream,
45
52
    record_to_fulltext_bytes,
46
53
    )
82
89
            recreate_search trusts that clients will look for missing things
83
90
            they expected and get it from elsewhere.
84
91
        """
 
92
        if search_bytes == 'everything':
 
93
            return vf_search.EverythingResult(repository), None
85
94
        lines = search_bytes.split('\n')
86
95
        if lines[0] == 'ancestry-of':
87
96
            heads = lines[1:]
88
 
            search_result = graph.PendingAncestryResult(heads, repository)
 
97
            search_result = vf_search.PendingAncestryResult(heads, repository)
89
98
            return search_result, None
90
99
        elif lines[0] == 'search':
91
100
            return self.recreate_search_from_recipe(repository, lines[1:],
115
124
                except StopIteration:
116
125
                    break
117
126
                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):
 
127
            (started_keys, excludes, included_keys) = search.get_state()
 
128
            if (not discard_excess and len(included_keys) != revision_count):
121
129
                # we got back a different amount of data than expected, this
122
130
                # gets reported as NoSuchRevision, because less revisions
123
131
                # indicates missing revisions, and more should never happen as
124
132
                # the excludes list considers ghosts and ensures that ghost
125
133
                # filling races are not a problem.
126
134
                return (None, FailedSmartServerResponse(('NoSuchRevision',)))
 
135
            search_result = vf_search.SearchResult(started_keys, excludes,
 
136
                len(included_keys), included_keys)
127
137
            return (search_result, None)
128
138
        finally:
129
139
            repository.unlock()
141
151
            repository.unlock()
142
152
 
143
153
 
 
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
 
144
164
class SmartServerRepositoryGetParentMap(SmartServerRepositoryRequest):
145
165
    """Bzr 1.2+ - get parent data for revisions during a graph search."""
146
166
 
179
199
        finally:
180
200
            repository.unlock()
181
201
 
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()
 
202
    def _expand_requested_revs(self, repo_graph, revision_ids, client_seen_revs,
 
203
                               include_missing, max_size=65536):
200
204
        result = {}
201
205
        queried_revs = set()
202
 
        size_so_far = 0
 
206
        estimator = estimate_compressed_size.ZLibEstimator(max_size)
203
207
        next_revs = revision_ids
204
208
        first_loop_done = False
205
209
        while next_revs:
227
231
                    # add parents to the result
228
232
                    result[encoded_id] = parents
229
233
                    # Approximate the serialized cost of this revision_id.
230
 
                    size_so_far += 2 + len(encoded_id) + sum(map(len, parents))
 
234
                    line = '%s %s\n' % (encoded_id, ' '.join(parents))
 
235
                    estimator.add_content(line)
231
236
            # get all the directly asked for parents, and then flesh out to
232
237
            # 64K (compressed) or so. We do one level of depth at a time to
233
238
            # stay in sync with the client. The 250000 magic number is
234
239
            # estimated compression ratio taken from bzr.dev itself.
235
 
            if self.no_extra_results or (
236
 
                first_loop_done and size_so_far > 250000):
 
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))
237
244
                next_revs = set()
238
245
                break
239
246
            # don't query things we've already queried
240
 
            next_revs.difference_update(queried_revs)
 
247
            next_revs = next_revs.difference(queried_revs)
241
248
            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)
242
271
 
243
272
        # sorting trivially puts lexographically similar revision ids together.
244
273
        # Compression FTW.
 
274
        lines = []
245
275
        for revision, parents in sorted(result.items()):
246
276
            lines.append(' '.join((revision, ) + tuple(parents)))
247
277
 
312
342
                ('history-incomplete', earliest_revno, earliest_revid))
313
343
 
314
344
 
 
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
 
315
359
class SmartServerRequestHasRevision(SmartServerRepositoryRequest):
316
360
 
317
361
    def do_repository_request(self, repository, revision_id):
319
363
 
320
364
        :param repository: The repository to query in.
321
365
        :param revision_id: The utf8 encoded revision_id to lookup.
322
 
        :return: A smart server response of ('ok', ) if the revision is
323
 
            present.
 
366
        :return: A smart server response of ('yes', ) if the revision is
 
367
            present. ('no', ) if it is missing.
324
368
        """
325
369
        if repository.has_revision(revision_id):
326
370
            return SuccessfulSmartServerResponse(('yes', ))
328
372
            return SuccessfulSmartServerResponse(('no', ))
329
373
 
330
374
 
 
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
 
331
399
class SmartServerRepositoryGatherStats(SmartServerRepositoryRequest):
332
400
 
333
401
    def do_repository_request(self, repository, revid, committers):
353
421
            decoded_committers = True
354
422
        else:
355
423
            decoded_committers = None
356
 
        stats = repository.gather_stats(decoded_revision_id, decoded_committers)
 
424
        try:
 
425
            stats = repository.gather_stats(decoded_revision_id,
 
426
                decoded_committers)
 
427
        except errors.NoSuchRevision:
 
428
            return FailedSmartServerResponse(('nosuchrevision', revid))
357
429
 
358
430
        body = ''
359
431
        if stats.has_key('committers'):
370
442
        return SuccessfulSmartServerResponse(('ok', ), body)
371
443
 
372
444
 
 
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
 
373
467
class SmartServerRepositoryIsShared(SmartServerRepositoryRequest):
374
468
 
375
469
    def do_repository_request(self, repository):
385
479
            return SuccessfulSmartServerResponse(('no', ))
386
480
 
387
481
 
 
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
 
388
499
class SmartServerRepositoryLockWrite(SmartServerRepositoryRequest):
389
500
 
390
501
    def do_repository_request(self, repository, token=''):
392
503
        if token == '':
393
504
            token = None
394
505
        try:
395
 
            token = repository.lock_write(token=token)
 
506
            token = repository.lock_write(token=token).repository_token
396
507
        except errors.LockContention, e:
397
508
            return FailedSmartServerResponse(('LockContention',))
398
509
        except errors.UnlockableTransport:
413
524
    def do_repository_request(self, repository, to_network_name):
414
525
        """Get a stream for inserting into a to_format repository.
415
526
 
 
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
 
416
534
        :param repository: The repository to stream from.
417
535
        :param to_network_name: The network name of the format of the target
418
536
            repository.
490
608
 
491
609
 
492
610
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
    """
493
618
 
494
619
    def _should_fake_unknown(self):
495
620
        """Returns False; we don't need to workaround bugs in 1.19+ clients."""
505
630
        for record in substream:
506
631
            if record.storage_kind in ('chunked', 'fulltext'):
507
632
                serialised = record_to_fulltext_bytes(record)
508
 
            elif record.storage_kind == 'inventory-delta':
509
 
                serialised = record_to_inventory_delta_bytes(record)
510
633
            elif record.storage_kind == 'absent':
511
634
                raise ValueError("Absent factory for %s" % (record.key,))
512
635
            else:
544
667
    :ivar first_bytes: The first bytes to give the next NetworkRecordStream.
545
668
    """
546
669
 
547
 
    def __init__(self, byte_stream):
 
670
    def __init__(self, byte_stream, record_counter):
548
671
        """Create a _ByteStreamDecoder."""
549
672
        self.stream_decoder = pack.ContainerPushParser()
550
673
        self.current_type = None
551
674
        self.first_bytes = None
552
675
        self.byte_stream = byte_stream
 
676
        self._record_counter = record_counter
 
677
        self.key_count = 0
553
678
 
554
679
    def iter_stream_decoder(self):
555
680
        """Iterate the contents of the pack from stream_decoder."""
580
705
 
581
706
    def record_stream(self):
582
707
        """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
 
583
736
        self.seed_state()
584
 
        # Make and consume sub generators, one per substream type:
585
 
        while self.first_bytes is not None:
586
 
            substream = NetworkRecordStream(self.iter_substream_bytes())
587
 
            # after substream is fully consumed, self.current_type is set to
588
 
            # the next type, and self.first_bytes is set to the matching bytes.
589
 
            yield self.current_type, substream.read()
 
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()
590
751
 
591
752
    def seed_state(self):
592
753
        """Prepare the _ByteStreamDecoder to decode from the pack stream."""
597
758
        list(self.iter_substream_bytes())
598
759
 
599
760
 
600
 
def _byte_stream_to_stream(byte_stream):
 
761
def _byte_stream_to_stream(byte_stream, record_counter=None):
601
762
    """Convert a byte stream into a format and a stream.
602
763
 
603
764
    :param byte_stream: A bytes iterator, as output by _stream_to_byte_stream.
604
765
    :return: (RepositoryFormat, stream_generator)
605
766
    """
606
 
    decoder = _ByteStreamDecoder(byte_stream)
 
767
    decoder = _ByteStreamDecoder(byte_stream, record_counter)
607
768
    for bytes in byte_stream:
608
769
        decoder.stream_decoder.accept_bytes(bytes)
609
770
        for record in decoder.stream_decoder.read_pending_records(max=1):
624
785
        return SuccessfulSmartServerResponse(('ok',))
625
786
 
626
787
 
 
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
 
627
801
class SmartServerRepositorySetMakeWorkingTrees(SmartServerRepositoryRequest):
628
802
 
629
803
    def do_repository_request(self, repository, str_bool_new_value):
792
966
        self.do_insert_stream_request(repository, resume_tokens)
793
967
 
794
968
 
 
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