/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 breezy/smart/repository.py

  • Committer: Jelmer Vernooij
  • Date: 2017-05-30 22:17:31 UTC
  • mto: This revision was merged to the branch mainline in revision 6642.
  • Revision ID: jelmer@jelmer.uk-20170530221731-k3djl8mbldt2gmoi
Fix doc generation with sphinx.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/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
 
22
import itertools
20
23
import os
21
 
import Queue
 
24
try:
 
25
    import queue
 
26
except ImportError:
 
27
    import Queue as queue
22
28
import sys
23
29
import tempfile
24
30
import threading
 
31
import zlib
25
32
 
26
 
from bzrlib import (
 
33
from .. import (
27
34
    bencode,
28
35
    errors,
29
 
    graph,
 
36
    estimate_compressed_size,
 
37
    inventory as _mod_inventory,
 
38
    inventory_delta,
30
39
    osutils,
31
40
    pack,
 
41
    trace,
32
42
    ui,
33
 
    versionedfile,
 
43
    vf_search,
34
44
    )
35
 
from bzrlib.bzrdir import BzrDir
36
 
from bzrlib.smart.request import (
 
45
from ..bzrdir import BzrDir
 
46
from ..sixish import (
 
47
    reraise,
 
48
)
 
49
from .request import (
37
50
    FailedSmartServerResponse,
38
51
    SmartServerRequest,
39
52
    SuccessfulSmartServerResponse,
40
53
    )
41
 
from bzrlib.repository import _strip_NULL_ghosts, network_format_registry
42
 
from bzrlib import revision as _mod_revision
43
 
from bzrlib.versionedfile import (
 
54
from ..repository import _strip_NULL_ghosts, network_format_registry
 
55
from .. import revision as _mod_revision
 
56
from ..versionedfile import (
 
57
    ChunkedContentFactory,
44
58
    NetworkRecordStream,
45
59
    record_to_fulltext_bytes,
46
60
    )
82
96
            recreate_search trusts that clients will look for missing things
83
97
            they expected and get it from elsewhere.
84
98
        """
 
99
        if search_bytes == 'everything':
 
100
            return vf_search.EverythingResult(repository), None
85
101
        lines = search_bytes.split('\n')
86
102
        if lines[0] == 'ancestry-of':
87
103
            heads = lines[1:]
88
 
            search_result = graph.PendingAncestryResult(heads, repository)
 
104
            search_result = vf_search.PendingAncestryResult(heads, repository)
89
105
            return search_result, None
90
106
        elif lines[0] == 'search':
91
107
            return self.recreate_search_from_recipe(repository, lines[1:],
111
127
                start_keys)
112
128
            while True:
113
129
                try:
114
 
                    next_revs = search.next()
 
130
                    next_revs = next(search)
115
131
                except StopIteration:
116
132
                    break
117
133
                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):
 
134
            (started_keys, excludes, included_keys) = search.get_state()
 
135
            if (not discard_excess and len(included_keys) != revision_count):
121
136
                # we got back a different amount of data than expected, this
122
137
                # gets reported as NoSuchRevision, because less revisions
123
138
                # indicates missing revisions, and more should never happen as
124
139
                # the excludes list considers ghosts and ensures that ghost
125
140
                # filling races are not a problem.
126
141
                return (None, FailedSmartServerResponse(('NoSuchRevision',)))
 
142
            search_result = vf_search.SearchResult(started_keys, excludes,
 
143
                len(included_keys), included_keys)
127
144
            return (search_result, None)
128
145
        finally:
129
146
            repository.unlock()
141
158
            repository.unlock()
142
159
 
143
160
 
 
161
class SmartServerRepositoryBreakLock(SmartServerRepositoryRequest):
 
162
    """Break a repository lock."""
 
163
 
 
164
    def do_repository_request(self, repository):
 
165
        repository.break_lock()
 
166
        return SuccessfulSmartServerResponse(('ok', ))
 
167
 
 
168
 
 
169
_lsprof_count = 0
 
170
 
144
171
class SmartServerRepositoryGetParentMap(SmartServerRepositoryRequest):
145
172
    """Bzr 1.2+ - get parent data for revisions during a graph search."""
146
173
 
179
206
        finally:
180
207
            repository.unlock()
181
208
 
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()
 
209
    def _expand_requested_revs(self, repo_graph, revision_ids, client_seen_revs,
 
210
                               include_missing, max_size=65536):
200
211
        result = {}
201
212
        queried_revs = set()
202
 
        size_so_far = 0
 
213
        estimator = estimate_compressed_size.ZLibEstimator(max_size)
203
214
        next_revs = revision_ids
204
215
        first_loop_done = False
205
216
        while next_revs:
227
238
                    # add parents to the result
228
239
                    result[encoded_id] = parents
229
240
                    # Approximate the serialized cost of this revision_id.
230
 
                    size_so_far += 2 + len(encoded_id) + sum(map(len, parents))
 
241
                    line = '%s %s\n' % (encoded_id, ' '.join(parents))
 
242
                    estimator.add_content(line)
231
243
            # get all the directly asked for parents, and then flesh out to
232
244
            # 64K (compressed) or so. We do one level of depth at a time to
233
245
            # stay in sync with the client. The 250000 magic number is
234
246
            # estimated compression ratio taken from bzr.dev itself.
235
 
            if self.no_extra_results or (
236
 
                first_loop_done and size_so_far > 250000):
 
247
            if self.no_extra_results or (first_loop_done and estimator.full()):
 
248
                trace.mutter('size: %d, z_size: %d'
 
249
                             % (estimator._uncompressed_size_added,
 
250
                                estimator._compressed_size_added))
237
251
                next_revs = set()
238
252
                break
239
253
            # don't query things we've already queried
240
 
            next_revs.difference_update(queried_revs)
 
254
            next_revs = next_revs.difference(queried_revs)
241
255
            first_loop_done = True
 
256
        return result
 
257
 
 
258
    def _do_repository_request(self, body_bytes):
 
259
        repository = self._repository
 
260
        revision_ids = set(self._revision_ids)
 
261
        include_missing = 'include-missing:' in revision_ids
 
262
        if include_missing:
 
263
            revision_ids.remove('include-missing:')
 
264
        body_lines = body_bytes.split('\n')
 
265
        search_result, error = self.recreate_search_from_recipe(
 
266
            repository, body_lines)
 
267
        if error is not None:
 
268
            return error
 
269
        # TODO might be nice to start up the search again; but thats not
 
270
        # written or tested yet.
 
271
        client_seen_revs = set(search_result.get_keys())
 
272
        # Always include the requested ids.
 
273
        client_seen_revs.difference_update(revision_ids)
 
274
 
 
275
        repo_graph = repository.get_graph()
 
276
        result = self._expand_requested_revs(repo_graph, revision_ids,
 
277
                                             client_seen_revs, include_missing)
242
278
 
243
279
        # sorting trivially puts lexographically similar revision ids together.
244
280
        # Compression FTW.
 
281
        lines = []
245
282
        for revision, parents in sorted(result.items()):
246
283
            lines.append(' '.join((revision, ) + tuple(parents)))
247
284
 
271
308
        else:
272
309
            search_ids = repository.all_revision_ids()
273
310
        search = graph._make_breadth_first_searcher(search_ids)
274
 
        transitive_ids = set()
275
 
        map(transitive_ids.update, list(search))
 
311
        transitive_ids = set(itertools.chain.from_iterable(search))
276
312
        parent_map = graph.get_parent_map(transitive_ids)
277
313
        revision_graph = _strip_NULL_ghosts(parent_map)
278
314
        if revision_id and revision_id not in revision_graph:
297
333
        """
298
334
        try:
299
335
            found_flag, result = repository.get_rev_id_for_revno(revno, known_pair)
300
 
        except errors.RevisionNotPresent, err:
 
336
        except errors.RevisionNotPresent as err:
301
337
            if err.revision_id != known_pair[1]:
302
338
                raise AssertionError(
303
339
                    'get_rev_id_for_revno raised RevisionNotPresent for '
312
348
                ('history-incomplete', earliest_revno, earliest_revid))
313
349
 
314
350
 
 
351
class SmartServerRepositoryGetSerializerFormat(SmartServerRepositoryRequest):
 
352
 
 
353
    def do_repository_request(self, repository):
 
354
        """Return the serializer format for this repository.
 
355
 
 
356
        New in 2.5.0.
 
357
 
 
358
        :param repository: The repository to query
 
359
        :return: A smart server response ('ok', FORMAT)
 
360
        """
 
361
        serializer = repository.get_serializer_format()
 
362
        return SuccessfulSmartServerResponse(('ok', serializer))
 
363
 
 
364
 
315
365
class SmartServerRequestHasRevision(SmartServerRepositoryRequest):
316
366
 
317
367
    def do_repository_request(self, repository, revision_id):
319
369
 
320
370
        :param repository: The repository to query in.
321
371
        :param revision_id: The utf8 encoded revision_id to lookup.
322
 
        :return: A smart server response of ('ok', ) if the revision is
323
 
            present.
 
372
        :return: A smart server response of ('yes', ) if the revision is
 
373
            present. ('no', ) if it is missing.
324
374
        """
325
375
        if repository.has_revision(revision_id):
326
376
            return SuccessfulSmartServerResponse(('yes', ))
328
378
            return SuccessfulSmartServerResponse(('no', ))
329
379
 
330
380
 
 
381
class SmartServerRequestHasSignatureForRevisionId(
 
382
        SmartServerRepositoryRequest):
 
383
 
 
384
    def do_repository_request(self, repository, revision_id):
 
385
        """Return ok if a signature is present for a revision.
 
386
 
 
387
        Introduced in bzr 2.5.0.
 
388
 
 
389
        :param repository: The repository to query in.
 
390
        :param revision_id: The utf8 encoded revision_id to lookup.
 
391
        :return: A smart server response of ('yes', ) if a
 
392
            signature for the revision is present,
 
393
            ('no', ) if it is missing.
 
394
        """
 
395
        try:
 
396
            if repository.has_signature_for_revision_id(revision_id):
 
397
                return SuccessfulSmartServerResponse(('yes', ))
 
398
            else:
 
399
                return SuccessfulSmartServerResponse(('no', ))
 
400
        except errors.NoSuchRevision:
 
401
            return FailedSmartServerResponse(
 
402
                ('nosuchrevision', revision_id))
 
403
 
 
404
 
331
405
class SmartServerRepositoryGatherStats(SmartServerRepositoryRequest):
332
406
 
333
407
    def do_repository_request(self, repository, revid, committers):
353
427
            decoded_committers = True
354
428
        else:
355
429
            decoded_committers = None
356
 
        stats = repository.gather_stats(decoded_revision_id, decoded_committers)
 
430
        try:
 
431
            stats = repository.gather_stats(decoded_revision_id,
 
432
                decoded_committers)
 
433
        except errors.NoSuchRevision:
 
434
            return FailedSmartServerResponse(('nosuchrevision', revid))
357
435
 
358
436
        body = ''
359
 
        if stats.has_key('committers'):
 
437
        if 'committers' in stats:
360
438
            body += 'committers: %d\n' % stats['committers']
361
 
        if stats.has_key('firstrev'):
 
439
        if 'firstrev' in stats:
362
440
            body += 'firstrev: %.3f %d\n' % stats['firstrev']
363
 
        if stats.has_key('latestrev'):
 
441
        if 'latestrev' in stats:
364
442
             body += 'latestrev: %.3f %d\n' % stats['latestrev']
365
 
        if stats.has_key('revisions'):
 
443
        if 'revisions' in stats:
366
444
            body += 'revisions: %d\n' % stats['revisions']
367
 
        if stats.has_key('size'):
 
445
        if 'size' in stats:
368
446
            body += 'size: %d\n' % stats['size']
369
447
 
370
448
        return SuccessfulSmartServerResponse(('ok', ), body)
371
449
 
372
450
 
 
451
class SmartServerRepositoryGetRevisionSignatureText(
 
452
        SmartServerRepositoryRequest):
 
453
    """Return the signature text of a revision.
 
454
 
 
455
    New in 2.5.
 
456
    """
 
457
 
 
458
    def do_repository_request(self, repository, revision_id):
 
459
        """Return the result of repository.get_signature_text().
 
460
 
 
461
        :param repository: The repository to query in.
 
462
        :return: A smart server response of with the signature text as
 
463
            body.
 
464
        """
 
465
        try:
 
466
            text = repository.get_signature_text(revision_id)
 
467
        except errors.NoSuchRevision as err:
 
468
            return FailedSmartServerResponse(
 
469
                ('nosuchrevision', err.revision))
 
470
        return SuccessfulSmartServerResponse(('ok', ), text)
 
471
 
 
472
 
373
473
class SmartServerRepositoryIsShared(SmartServerRepositoryRequest):
374
474
 
375
475
    def do_repository_request(self, repository):
385
485
            return SuccessfulSmartServerResponse(('no', ))
386
486
 
387
487
 
 
488
class SmartServerRepositoryMakeWorkingTrees(SmartServerRepositoryRequest):
 
489
 
 
490
    def do_repository_request(self, repository):
 
491
        """Return the result of repository.make_working_trees().
 
492
 
 
493
        Introduced in bzr 2.5.0.
 
494
 
 
495
        :param repository: The repository to query in.
 
496
        :return: A smart server response of ('yes', ) if the repository uses
 
497
            working trees, and ('no', ) if it is not.
 
498
        """
 
499
        if repository.make_working_trees():
 
500
            return SuccessfulSmartServerResponse(('yes', ))
 
501
        else:
 
502
            return SuccessfulSmartServerResponse(('no', ))
 
503
 
 
504
 
388
505
class SmartServerRepositoryLockWrite(SmartServerRepositoryRequest):
389
506
 
390
507
    def do_repository_request(self, repository, token=''):
392
509
        if token == '':
393
510
            token = None
394
511
        try:
395
 
            token = repository.lock_write(token=token)
396
 
        except errors.LockContention, e:
 
512
            token = repository.lock_write(token=token).repository_token
 
513
        except errors.LockContention as e:
397
514
            return FailedSmartServerResponse(('LockContention',))
398
515
        except errors.UnlockableTransport:
399
516
            return FailedSmartServerResponse(('UnlockableTransport',))
400
 
        except errors.LockFailed, e:
 
517
        except errors.LockFailed as e:
401
518
            return FailedSmartServerResponse(('LockFailed',
402
519
                str(e.lock), str(e.why)))
403
520
        if token is not None:
413
530
    def do_repository_request(self, repository, to_network_name):
414
531
        """Get a stream for inserting into a to_format repository.
415
532
 
 
533
        The request body is 'search_bytes', a description of the revisions
 
534
        being requested.
 
535
 
 
536
        In 2.3 this verb added support for search_bytes == 'everything'.  Older
 
537
        implementations will respond with a BadSearch error, and clients should
 
538
        catch this and fallback appropriately.
 
539
 
416
540
        :param repository: The repository to stream from.
417
541
        :param to_network_name: The network name of the format of the target
418
542
            repository.
466
590
            source = repository._get_source(self._to_format)
467
591
            stream = source.get_stream(search_result)
468
592
        except Exception:
469
 
            exc_info = sys.exc_info()
470
593
            try:
471
594
                # On non-error, unlocking is done by the body stream handler.
472
595
                repository.unlock()
473
596
            finally:
474
 
                raise exc_info[0], exc_info[1], exc_info[2]
 
597
                raise
475
598
        return SuccessfulSmartServerResponse(('ok',),
476
599
            body_stream=self.body_stream(stream, repository))
477
600
 
480
603
        try:
481
604
            for bytes in byte_stream:
482
605
                yield bytes
483
 
        except errors.RevisionNotPresent, e:
 
606
        except errors.RevisionNotPresent as e:
484
607
            # This shouldn't be able to happen, but as we don't buffer
485
608
            # everything it can in theory happen.
486
609
            repository.unlock()
490
613
 
491
614
 
492
615
class SmartServerRepositoryGetStream_1_19(SmartServerRepositoryGetStream):
 
616
    """The same as Repository.get_stream, but will return stream CHK formats to
 
617
    clients.
 
618
 
 
619
    See SmartServerRepositoryGetStream._should_fake_unknown.
 
620
    
 
621
    New in 1.19.
 
622
    """
493
623
 
494
624
    def _should_fake_unknown(self):
495
625
        """Returns False; we don't need to workaround bugs in 1.19+ clients."""
505
635
        for record in substream:
506
636
            if record.storage_kind in ('chunked', 'fulltext'):
507
637
                serialised = record_to_fulltext_bytes(record)
508
 
            elif record.storage_kind == 'inventory-delta':
509
 
                serialised = record_to_inventory_delta_bytes(record)
510
638
            elif record.storage_kind == 'absent':
511
639
                raise ValueError("Absent factory for %s" % (record.key,))
512
640
            else:
544
672
    :ivar first_bytes: The first bytes to give the next NetworkRecordStream.
545
673
    """
546
674
 
547
 
    def __init__(self, byte_stream):
 
675
    def __init__(self, byte_stream, record_counter):
548
676
        """Create a _ByteStreamDecoder."""
549
677
        self.stream_decoder = pack.ContainerPushParser()
550
678
        self.current_type = None
551
679
        self.first_bytes = None
552
680
        self.byte_stream = byte_stream
 
681
        self._record_counter = record_counter
 
682
        self.key_count = 0
553
683
 
554
684
    def iter_stream_decoder(self):
555
685
        """Iterate the contents of the pack from stream_decoder."""
580
710
 
581
711
    def record_stream(self):
582
712
        """Yield substream_type, substream from the byte stream."""
 
713
        def wrap_and_count(pb, rc, substream):
 
714
            """Yield records from stream while showing progress."""
 
715
            counter = 0
 
716
            if rc:
 
717
                if self.current_type != 'revisions' and self.key_count != 0:
 
718
                    # As we know the number of revisions now (in self.key_count)
 
719
                    # we can setup and use record_counter (rc).
 
720
                    if not rc.is_initialized():
 
721
                        rc.setup(self.key_count, self.key_count)
 
722
            for record in substream.read():
 
723
                if rc:
 
724
                    if rc.is_initialized() and counter == rc.STEP:
 
725
                        rc.increment(counter)
 
726
                        pb.update('Estimate', rc.current, rc.max)
 
727
                        counter = 0
 
728
                    if self.current_type == 'revisions':
 
729
                        # Total records is proportional to number of revs
 
730
                        # to fetch. With remote, we used self.key_count to
 
731
                        # track the number of revs. Once we have the revs
 
732
                        # counts in self.key_count, the progress bar changes
 
733
                        # from 'Estimating..' to 'Estimate' above.
 
734
                        self.key_count += 1
 
735
                        if counter == rc.STEP:
 
736
                            pb.update('Estimating..', self.key_count)
 
737
                            counter = 0
 
738
                counter += 1
 
739
                yield record
 
740
 
583
741
        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()
 
742
        pb = ui.ui_factory.nested_progress_bar()
 
743
        rc = self._record_counter
 
744
        try:
 
745
            # Make and consume sub generators, one per substream type:
 
746
            while self.first_bytes is not None:
 
747
                substream = NetworkRecordStream(self.iter_substream_bytes())
 
748
                # after substream is fully consumed, self.current_type is set
 
749
                # to the next type, and self.first_bytes is set to the matching
 
750
                # bytes.
 
751
                yield self.current_type, wrap_and_count(pb, rc, substream)
 
752
        finally:
 
753
            if rc:
 
754
                pb.update('Done', rc.max, rc.max)
 
755
            pb.finished()
590
756
 
591
757
    def seed_state(self):
592
758
        """Prepare the _ByteStreamDecoder to decode from the pack stream."""
597
763
        list(self.iter_substream_bytes())
598
764
 
599
765
 
600
 
def _byte_stream_to_stream(byte_stream):
 
766
def _byte_stream_to_stream(byte_stream, record_counter=None):
601
767
    """Convert a byte stream into a format and a stream.
602
768
 
603
769
    :param byte_stream: A bytes iterator, as output by _stream_to_byte_stream.
604
770
    :return: (RepositoryFormat, stream_generator)
605
771
    """
606
 
    decoder = _ByteStreamDecoder(byte_stream)
 
772
    decoder = _ByteStreamDecoder(byte_stream, record_counter)
607
773
    for bytes in byte_stream:
608
774
        decoder.stream_decoder.accept_bytes(bytes)
609
775
        for record in decoder.stream_decoder.read_pending_records(max=1):
617
783
    def do_repository_request(self, repository, token):
618
784
        try:
619
785
            repository.lock_write(token=token)
620
 
        except errors.TokenMismatch, e:
 
786
        except errors.TokenMismatch as e:
621
787
            return FailedSmartServerResponse(('TokenMismatch',))
622
788
        repository.dont_leave_lock_in_place()
623
789
        repository.unlock()
624
790
        return SuccessfulSmartServerResponse(('ok',))
625
791
 
626
792
 
 
793
class SmartServerRepositoryGetPhysicalLockStatus(SmartServerRepositoryRequest):
 
794
    """Get the physical lock status for a repository.
 
795
 
 
796
    New in 2.5.
 
797
    """
 
798
 
 
799
    def do_repository_request(self, repository):
 
800
        if repository.get_physical_lock_status():
 
801
            return SuccessfulSmartServerResponse(('yes', ))
 
802
        else:
 
803
            return SuccessfulSmartServerResponse(('no', ))
 
804
 
 
805
 
627
806
class SmartServerRepositorySetMakeWorkingTrees(SmartServerRepositoryRequest):
628
807
 
629
808
    def do_repository_request(self, repository, str_bool_new_value):
713
892
        tokens = [token for token in resume_tokens.split(' ') if token]
714
893
        self.tokens = tokens
715
894
        self.repository = repository
716
 
        self.queue = Queue.Queue()
 
895
        self.queue = queue.Queue()
717
896
        self.insert_thread = threading.Thread(target=self._inserter_thread)
718
897
        self.insert_thread.start()
719
898
 
744
923
        if self.insert_thread is not None:
745
924
            self.insert_thread.join()
746
925
        if not self.insert_ok:
747
 
            exc_info = self.insert_exception
748
 
            raise exc_info[0], exc_info[1], exc_info[2]
 
926
            try:
 
927
                reraise(*self.insert_exception)
 
928
            finally:
 
929
                del self.insert_exception
749
930
        write_group_tokens, missing_keys = self.insert_result
750
931
        if write_group_tokens or missing_keys:
751
932
            # bzip needed? missing keys should typically be a small set.
792
973
        self.do_insert_stream_request(repository, resume_tokens)
793
974
 
794
975
 
 
976
class SmartServerRepositoryAddSignatureText(SmartServerRepositoryRequest):
 
977
    """Add a revision signature text.
 
978
 
 
979
    New in 2.5.
 
980
    """
 
981
 
 
982
    def do_repository_request(self, repository, lock_token, revision_id,
 
983
            *write_group_tokens):
 
984
        """Add a revision signature text.
 
985
 
 
986
        :param repository: Repository to operate on
 
987
        :param lock_token: Lock token
 
988
        :param revision_id: Revision for which to add signature
 
989
        :param write_group_tokens: Write group tokens
 
990
        """
 
991
        self._lock_token = lock_token
 
992
        self._revision_id = revision_id
 
993
        self._write_group_tokens = write_group_tokens
 
994
        return None
 
995
 
 
996
    def do_body(self, body_bytes):
 
997
        """Add a signature text.
 
998
 
 
999
        :param body_bytes: GPG signature text
 
1000
        :return: SuccessfulSmartServerResponse with arguments 'ok' and
 
1001
            the list of new write group tokens.
 
1002
        """
 
1003
        self._repository.lock_write(token=self._lock_token)
 
1004
        try:
 
1005
            self._repository.resume_write_group(self._write_group_tokens)
 
1006
            try:
 
1007
                self._repository.add_signature_text(self._revision_id,
 
1008
                    body_bytes)
 
1009
            finally:
 
1010
                new_write_group_tokens = self._repository.suspend_write_group()
 
1011
        finally:
 
1012
            self._repository.unlock()
 
1013
        return SuccessfulSmartServerResponse(
 
1014
            ('ok', ) + tuple(new_write_group_tokens))
 
1015
 
 
1016
 
 
1017
class SmartServerRepositoryStartWriteGroup(SmartServerRepositoryRequest):
 
1018
    """Start a write group.
 
1019
 
 
1020
    New in 2.5.
 
1021
    """
 
1022
 
 
1023
    def do_repository_request(self, repository, lock_token):
 
1024
        """Start a write group."""
 
1025
        repository.lock_write(token=lock_token)
 
1026
        try:
 
1027
            repository.start_write_group()
 
1028
            try:
 
1029
                tokens = repository.suspend_write_group()
 
1030
            except errors.UnsuspendableWriteGroup:
 
1031
                return FailedSmartServerResponse(('UnsuspendableWriteGroup',))
 
1032
        finally:
 
1033
            repository.unlock()
 
1034
        return SuccessfulSmartServerResponse(('ok', tokens))
 
1035
 
 
1036
 
 
1037
class SmartServerRepositoryCommitWriteGroup(SmartServerRepositoryRequest):
 
1038
    """Commit a write group.
 
1039
 
 
1040
    New in 2.5.
 
1041
    """
 
1042
 
 
1043
    def do_repository_request(self, repository, lock_token,
 
1044
            write_group_tokens):
 
1045
        """Commit a write group."""
 
1046
        repository.lock_write(token=lock_token)
 
1047
        try:
 
1048
            try:
 
1049
                repository.resume_write_group(write_group_tokens)
 
1050
            except errors.UnresumableWriteGroup as e:
 
1051
                return FailedSmartServerResponse(
 
1052
                    ('UnresumableWriteGroup', e.write_groups, e.reason))
 
1053
            try:
 
1054
                repository.commit_write_group()
 
1055
            except:
 
1056
                write_group_tokens = repository.suspend_write_group()
 
1057
                # FIXME JRV 2011-11-19: What if the write_group_tokens
 
1058
                # have changed?
 
1059
                raise
 
1060
        finally:
 
1061
            repository.unlock()
 
1062
        return SuccessfulSmartServerResponse(('ok', ))
 
1063
 
 
1064
 
 
1065
class SmartServerRepositoryAbortWriteGroup(SmartServerRepositoryRequest):
 
1066
    """Abort a write group.
 
1067
 
 
1068
    New in 2.5.
 
1069
    """
 
1070
 
 
1071
    def do_repository_request(self, repository, lock_token, write_group_tokens):
 
1072
        """Abort a write group."""
 
1073
        repository.lock_write(token=lock_token)
 
1074
        try:
 
1075
            try:
 
1076
                repository.resume_write_group(write_group_tokens)
 
1077
            except errors.UnresumableWriteGroup as e:
 
1078
                return FailedSmartServerResponse(
 
1079
                    ('UnresumableWriteGroup', e.write_groups, e.reason))
 
1080
                repository.abort_write_group()
 
1081
        finally:
 
1082
            repository.unlock()
 
1083
        return SuccessfulSmartServerResponse(('ok', ))
 
1084
 
 
1085
 
 
1086
class SmartServerRepositoryCheckWriteGroup(SmartServerRepositoryRequest):
 
1087
    """Check that a write group is still valid.
 
1088
 
 
1089
    New in 2.5.
 
1090
    """
 
1091
 
 
1092
    def do_repository_request(self, repository, lock_token, write_group_tokens):
 
1093
        """Abort a write group."""
 
1094
        repository.lock_write(token=lock_token)
 
1095
        try:
 
1096
            try:
 
1097
                repository.resume_write_group(write_group_tokens)
 
1098
            except errors.UnresumableWriteGroup as e:
 
1099
                return FailedSmartServerResponse(
 
1100
                    ('UnresumableWriteGroup', e.write_groups, e.reason))
 
1101
            else:
 
1102
                repository.suspend_write_group()
 
1103
        finally:
 
1104
            repository.unlock()
 
1105
        return SuccessfulSmartServerResponse(('ok', ))
 
1106
 
 
1107
 
 
1108
class SmartServerRepositoryAllRevisionIds(SmartServerRepositoryRequest):
 
1109
    """Retrieve all of the revision ids in a repository.
 
1110
 
 
1111
    New in 2.5.
 
1112
    """
 
1113
 
 
1114
    def do_repository_request(self, repository):
 
1115
        revids = repository.all_revision_ids()
 
1116
        return SuccessfulSmartServerResponse(("ok", ), "\n".join(revids))
 
1117
 
 
1118
 
 
1119
class SmartServerRepositoryReconcile(SmartServerRepositoryRequest):
 
1120
    """Reconcile a repository.
 
1121
 
 
1122
    New in 2.5.
 
1123
    """
 
1124
 
 
1125
    def do_repository_request(self, repository, lock_token):
 
1126
        try:
 
1127
            repository.lock_write(token=lock_token)
 
1128
        except errors.TokenLockingNotSupported as e:
 
1129
            return FailedSmartServerResponse(
 
1130
                ('TokenLockingNotSupported', ))
 
1131
        try:
 
1132
            reconciler = repository.reconcile()
 
1133
        finally:
 
1134
            repository.unlock()
 
1135
        body = [
 
1136
            "garbage_inventories: %d\n" % reconciler.garbage_inventories,
 
1137
            "inconsistent_parents: %d\n" % reconciler.inconsistent_parents,
 
1138
            ]
 
1139
        return SuccessfulSmartServerResponse(('ok', ), "".join(body))
 
1140
 
 
1141
 
 
1142
class SmartServerRepositoryPack(SmartServerRepositoryRequest):
 
1143
    """Pack a repository.
 
1144
 
 
1145
    New in 2.5.
 
1146
    """
 
1147
 
 
1148
    def do_repository_request(self, repository, lock_token, clean_obsolete_packs):
 
1149
        self._repository = repository
 
1150
        self._lock_token = lock_token
 
1151
        if clean_obsolete_packs == 'True':
 
1152
            self._clean_obsolete_packs = True
 
1153
        else:
 
1154
            self._clean_obsolete_packs = False
 
1155
        return None
 
1156
 
 
1157
    def do_body(self, body_bytes):
 
1158
        if body_bytes == "":
 
1159
            hint = None
 
1160
        else:
 
1161
            hint = body_bytes.splitlines()
 
1162
        self._repository.lock_write(token=self._lock_token)
 
1163
        try:
 
1164
            self._repository.pack(hint, self._clean_obsolete_packs)
 
1165
        finally:
 
1166
            self._repository.unlock()
 
1167
        return SuccessfulSmartServerResponse(("ok", ), )
 
1168
 
 
1169
 
 
1170
class SmartServerRepositoryIterFilesBytes(SmartServerRepositoryRequest):
 
1171
    """Iterate over the contents of files.
 
1172
 
 
1173
    The client sends a list of desired files to stream, one
 
1174
    per line, and as tuples of file id and revision, separated by
 
1175
    \0.
 
1176
 
 
1177
    The server replies with a stream. Each entry is preceded by a header,
 
1178
    which can either be:
 
1179
 
 
1180
    * "ok\x00IDX\n" where IDX is the index of the entry in the desired files
 
1181
        list sent by the client. This header is followed by the contents of
 
1182
        the file, bzip2-compressed.
 
1183
    * "absent\x00FILEID\x00REVISION\x00IDX" to indicate a text is missing.
 
1184
        The client can then raise an appropriate RevisionNotPresent error
 
1185
        or check its fallback repositories.
 
1186
 
 
1187
    New in 2.5.
 
1188
    """
 
1189
 
 
1190
    def body_stream(self, repository, desired_files):
 
1191
        self._repository.lock_read()
 
1192
        try:
 
1193
            text_keys = {}
 
1194
            for i, key in enumerate(desired_files):
 
1195
                text_keys[key] = i
 
1196
            for record in repository.texts.get_record_stream(text_keys,
 
1197
                    'unordered', True):
 
1198
                identifier = text_keys[record.key]
 
1199
                if record.storage_kind == 'absent':
 
1200
                    yield "absent\0%s\0%s\0%d\n" % (record.key[0],
 
1201
                        record.key[1], identifier)
 
1202
                    # FIXME: Way to abort early?
 
1203
                    continue
 
1204
                yield "ok\0%d\n" % identifier
 
1205
                compressor = zlib.compressobj()
 
1206
                for bytes in record.get_bytes_as('chunked'):
 
1207
                    data = compressor.compress(bytes)
 
1208
                    if data:
 
1209
                        yield data
 
1210
                data = compressor.flush()
 
1211
                if data:
 
1212
                    yield data
 
1213
        finally:
 
1214
            self._repository.unlock()
 
1215
 
 
1216
    def do_body(self, body_bytes):
 
1217
        desired_files = [
 
1218
            tuple(l.split("\0")) for l in body_bytes.splitlines()]
 
1219
        return SuccessfulSmartServerResponse(('ok', ),
 
1220
            body_stream=self.body_stream(self._repository, desired_files))
 
1221
 
 
1222
    def do_repository_request(self, repository):
 
1223
        # Signal that we want a body
 
1224
        return None
 
1225
 
 
1226
 
 
1227
class SmartServerRepositoryIterRevisions(SmartServerRepositoryRequest):
 
1228
    """Stream a list of revisions.
 
1229
 
 
1230
    The client sends a list of newline-separated revision ids in the
 
1231
    body of the request and the server replies with the serializer format,
 
1232
    and a stream of bzip2-compressed revision texts (using the specified
 
1233
    serializer format).
 
1234
 
 
1235
    Any revisions the server does not have are omitted from the stream.
 
1236
 
 
1237
    New in 2.5.
 
1238
    """
 
1239
 
 
1240
    def do_repository_request(self, repository):
 
1241
        self._repository = repository
 
1242
        # Signal there is a body
 
1243
        return None
 
1244
 
 
1245
    def do_body(self, body_bytes):
 
1246
        revision_ids = body_bytes.split("\n")
 
1247
        return SuccessfulSmartServerResponse(
 
1248
            ('ok', self._repository.get_serializer_format()),
 
1249
            body_stream=self.body_stream(self._repository, revision_ids))
 
1250
 
 
1251
    def body_stream(self, repository, revision_ids):
 
1252
        self._repository.lock_read()
 
1253
        try:
 
1254
            for record in repository.revisions.get_record_stream(
 
1255
                [(revid,) for revid in revision_ids], 'unordered', True):
 
1256
                if record.storage_kind == 'absent':
 
1257
                    continue
 
1258
                yield zlib.compress(record.get_bytes_as('fulltext'))
 
1259
        finally:
 
1260
            self._repository.unlock()
 
1261
 
 
1262
 
 
1263
class SmartServerRepositoryGetInventories(SmartServerRepositoryRequest):
 
1264
    """Get the inventory deltas for a set of revision ids.
 
1265
 
 
1266
    This accepts a list of revision ids, and then sends a chain
 
1267
    of deltas for the inventories of those revisions. The first
 
1268
    revision will be empty.
 
1269
 
 
1270
    The server writes back zlibbed serialized inventory deltas,
 
1271
    in the ordering specified. The base for each delta is the
 
1272
    inventory generated by the previous delta.
 
1273
 
 
1274
    New in 2.5.
 
1275
    """
 
1276
 
 
1277
    def _inventory_delta_stream(self, repository, ordering, revids):
 
1278
        prev_inv = _mod_inventory.Inventory(root_id=None,
 
1279
            revision_id=_mod_revision.NULL_REVISION)
 
1280
        serializer = inventory_delta.InventoryDeltaSerializer(
 
1281
            repository.supports_rich_root(),
 
1282
            repository._format.supports_tree_reference)
 
1283
        repository.lock_read()
 
1284
        try:
 
1285
            for inv, revid in repository._iter_inventories(revids, ordering):
 
1286
                if inv is None:
 
1287
                    continue
 
1288
                inv_delta = inv._make_delta(prev_inv)
 
1289
                lines = serializer.delta_to_lines(
 
1290
                    prev_inv.revision_id, inv.revision_id, inv_delta)
 
1291
                yield ChunkedContentFactory(inv.revision_id, None, None, lines)
 
1292
                prev_inv = inv
 
1293
        finally:
 
1294
            repository.unlock()
 
1295
 
 
1296
    def body_stream(self, repository, ordering, revids):
 
1297
        substream = self._inventory_delta_stream(repository,
 
1298
            ordering, revids)
 
1299
        return _stream_to_byte_stream([('inventory-deltas', substream)],
 
1300
            repository._format)
 
1301
 
 
1302
    def do_body(self, body_bytes):
 
1303
        return SuccessfulSmartServerResponse(('ok', ),
 
1304
            body_stream=self.body_stream(self._repository, self._ordering,
 
1305
                body_bytes.splitlines()))
 
1306
 
 
1307
    def do_repository_request(self, repository, ordering):
 
1308
        if ordering == 'unordered':
 
1309
            # inventory deltas for a topologically sorted stream
 
1310
            # are likely to be smaller
 
1311
            ordering = 'topological'
 
1312
        self._ordering = ordering
 
1313
        # Signal that we want a body
 
1314
        return None