← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/repository.py

  • Committer: Aaron Bentley
  • Date: 2009-06-19 21:16:31 UTC
  • mto: This revision was merged to the branch mainline in revision 4481.
  • Revision ID: aaron@aaronbentley.com-20090619211631-4fnkv2uui98xj7ux
Provide control over switch and shelver messaging.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/repository.py
1
 
# Copyright (C) 2006-2010 Canonical Ltd
 
1
# Copyright (C) 2006, 2007 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
20
20
import os
21
21
import Queue
22
22
import sys
 
23
import tarfile
23
24
import tempfile
24
25
import threading
25
26
 
29
30
    graph,
30
31
    osutils,
31
32
    pack,
32
 
    ui,
33
33
    )
34
34
from bzrlib.bzrdir import BzrDir
35
35
from bzrlib.smart.request import (
39
39
    )
40
40
from bzrlib.repository import _strip_NULL_ghosts, network_format_registry
41
41
from bzrlib import revision as _mod_revision
42
 
from bzrlib.versionedfile import (
43
 
    NetworkRecordStream,
44
 
    record_to_fulltext_bytes,
45
 
    )
 
42
from bzrlib.versionedfile import NetworkRecordStream, record_to_fulltext_bytes
46
43
 
47
44
 
48
45
class SmartServerRepositoryRequest(SmartServerRequest):
81
78
            recreate_search trusts that clients will look for missing things
82
79
            they expected and get it from elsewhere.
83
80
        """
84
 
        if search_bytes == 'everything':
85
 
            return graph.EverythingResult(repository), None
86
81
        lines = search_bytes.split('\n')
87
82
        if lines[0] == 'ancestry-of':
88
83
            heads = lines[1:]
393
388
        if token == '':
394
389
            token = None
395
390
        try:
396
 
            token = repository.lock_write(token=token).repository_token
 
391
            token = repository.lock_write(token=token)
397
392
        except errors.LockContention, e:
398
393
            return FailedSmartServerResponse(('LockContention',))
399
394
        except errors.UnlockableTransport:
414
409
    def do_repository_request(self, repository, to_network_name):
415
410
        """Get a stream for inserting into a to_format repository.
416
411
 
417
 
        The request body is 'search_bytes', a description of the revisions
418
 
        being requested.
419
 
 
420
 
        In 2.3 this verb added support for search_bytes == 'everything'.  Older
421
 
        implementations will respond with a BadSearch error, and clients should
422
 
        catch this and fallback appropriately.
423
 
 
424
412
        :param repository: The repository to stream from.
425
413
        :param to_network_name: The network name of the format of the target
426
414
            repository.
427
415
        """
428
416
        self._to_format = network_format_registry.get(to_network_name)
429
 
        if self._should_fake_unknown():
430
 
            return FailedSmartServerResponse(
431
 
                ('UnknownMethod', 'Repository.get_stream'))
432
417
        return None # Signal that we want a body.
433
418
 
434
 
    def _should_fake_unknown(self):
435
 
        """Return True if we should return UnknownMethod to the client.
436
 
        
437
 
        This is a workaround for bugs in pre-1.19 clients that claim to
438
 
        support receiving streams of CHK repositories.  The pre-1.19 client
439
 
        expects inventory records to be serialized in the format defined by
440
 
        to_network_name, but in pre-1.19 (at least) that format definition
441
 
        tries to use the xml5 serializer, which does not correctly handle
442
 
        rich-roots.  After 1.19 the client can also accept inventory-deltas
443
 
        (which avoids this issue), and those clients will use the
444
 
        Repository.get_stream_1.19 verb instead of this one.
445
 
        So: if this repository is CHK, and the to_format doesn't match,
446
 
        we should just fake an UnknownSmartMethod error so that the client
447
 
        will fallback to VFS, rather than sending it a stream we know it
448
 
        cannot handle.
449
 
        """
450
 
        from_format = self._repository._format
451
 
        to_format = self._to_format
452
 
        if not from_format.supports_chks:
453
 
            # Source not CHK: that's ok
454
 
            return False
455
 
        if (to_format.supports_chks and
456
 
            from_format.repository_class is to_format.repository_class and
457
 
            from_format._serializer == to_format._serializer):
458
 
            # Source is CHK, but target matches: that's ok
459
 
            # (e.g. 2a->2a, or CHK2->2a)
460
 
            return False
461
 
        # Source is CHK, and target is not CHK or incompatible CHK.  We can't
462
 
        # generate a compatible stream.
463
 
        return True
464
 
 
465
419
    def do_body(self, body_bytes):
466
420
        repository = self._repository
467
421
        repository.lock_read()
497
451
            repository.unlock()
498
452
 
499
453
 
500
 
class SmartServerRepositoryGetStream_1_19(SmartServerRepositoryGetStream):
501
 
    """The same as Repository.get_stream, but will return stream CHK formats to
502
 
    clients.
503
 
 
504
 
    See SmartServerRepositoryGetStream._should_fake_unknown.
505
 
    
506
 
    New in 1.19.
507
 
    """
508
 
 
509
 
    def _should_fake_unknown(self):
510
 
        """Returns False; we don't need to workaround bugs in 1.19+ clients."""
511
 
        return False
512
 
 
513
 
 
514
454
def _stream_to_byte_stream(stream, src_format):
515
455
    """Convert a record stream to a self delimited byte stream."""
516
456
    pack_writer = pack.ContainerSerialiser()
532
472
    yield pack_writer.end()
533
473
 
534
474
 
535
 
class _ByteStreamDecoder(object):
536
 
    """Helper for _byte_stream_to_stream.
537
 
 
538
 
    The expected usage of this class is via the function _byte_stream_to_stream
539
 
    which creates a _ByteStreamDecoder, pops off the stream format and then
540
 
    yields the output of record_stream(), the main entry point to
541
 
    _ByteStreamDecoder.
542
 
 
543
 
    Broadly this class has to unwrap two layers of iterators:
544
 
    (type, substream)
545
 
    (substream details)
546
 
 
547
 
    This is complicated by wishing to return type, iterator_for_type, but
548
 
    getting the data for iterator_for_type when we find out type: we can't
549
 
    simply pass a generator down to the NetworkRecordStream parser, instead
550
 
    we have a little local state to seed each NetworkRecordStream instance,
551
 
    and gather the type that we'll be yielding.
552
 
 
553
 
    :ivar byte_stream: The byte stream being decoded.
554
 
    :ivar stream_decoder: A pack parser used to decode the bytestream
555
 
    :ivar current_type: The current type, used to join adjacent records of the
556
 
        same type into a single stream.
557
 
    :ivar first_bytes: The first bytes to give the next NetworkRecordStream.
 
475
def _byte_stream_to_stream(byte_stream):
 
476
    """Convert a byte stream into a format and a stream.
 
477
 
 
478
    :param byte_stream: A bytes iterator, as output by _stream_to_byte_stream.
 
479
    :return: (RepositoryFormat, stream_generator)
558
480
    """
559
 
 
560
 
    def __init__(self, byte_stream, record_counter):
561
 
        """Create a _ByteStreamDecoder."""
562
 
        self.stream_decoder = pack.ContainerPushParser()
563
 
        self.current_type = None
564
 
        self.first_bytes = None
565
 
        self.byte_stream = byte_stream
566
 
        self._record_counter = record_counter
567
 
        self.key_count = 0
568
 
 
569
 
    def iter_stream_decoder(self):
570
 
        """Iterate the contents of the pack from stream_decoder."""
571
 
        # dequeue pending items
572
 
        for record in self.stream_decoder.read_pending_records():
573
 
            yield record
574
 
        # Pull bytes of the wire, decode them to records, yield those records.
575
 
        for bytes in self.byte_stream:
576
 
            self.stream_decoder.accept_bytes(bytes)
577
 
            for record in self.stream_decoder.read_pending_records():
578
 
                yield record
579
 
 
580
 
    def iter_substream_bytes(self):
581
 
        if self.first_bytes is not None:
582
 
            yield self.first_bytes
583
 
            # If we run out of pack records, single the outer layer to stop.
584
 
            self.first_bytes = None
585
 
        for record in self.iter_pack_records:
 
481
    stream_decoder = pack.ContainerPushParser()
 
482
    def record_stream():
 
483
        """Closure to return the substreams."""
 
484
        # May have fully parsed records already.
 
485
        for record in stream_decoder.read_pending_records():
586
486
            record_names, record_bytes = record
587
487
            record_name, = record_names
588
488
            substream_type = record_name[0]
589
 
            if substream_type != self.current_type:
590
 
                # end of a substream, seed the next substream.
591
 
                self.current_type = substream_type
592
 
                self.first_bytes = record_bytes
593
 
                return
594
 
            yield record_bytes
595
 
 
596
 
    def record_stream(self):
597
 
        """Yield substream_type, substream from the byte stream."""
598
 
        def wrap_and_count(pb, rc, substream):
599
 
            """Yield records from stream while showing progress."""
600
 
            counter = 0
601
 
            if rc:
602
 
                if self.current_type != 'revisions' and self.key_count != 0:
603
 
                    # As we know the number of revisions now (in self.key_count)
604
 
                    # we can setup and use record_counter (rc).
605
 
                    if not rc.is_initialized():
606
 
                        rc.setup(self.key_count, self.key_count)
607
 
            for record in substream.read():
608
 
                if rc:
609
 
                    if rc.is_initialized() and counter == rc.STEP:
610
 
                        rc.increment(counter)
611
 
                        pb.update('Estimate', rc.current, rc.max)
612
 
                        counter = 0
613
 
                    if self.current_type == 'revisions':
614
 
                        # Total records is proportional to number of revs
615
 
                        # to fetch. With remote, we used self.key_count to
616
 
                        # track the number of revs. Once we have the revs
617
 
                        # counts in self.key_count, the progress bar changes
618
 
                        # from 'Estimating..' to 'Estimate' above.
619
 
                        self.key_count += 1
620
 
                        if counter == rc.STEP:
621
 
                            pb.update('Estimating..', self.key_count)
622
 
                            counter = 0
623
 
                counter += 1
624
 
                yield record
625
 
 
626
 
        self.seed_state()
627
 
        pb = ui.ui_factory.nested_progress_bar()
628
 
        rc = self._record_counter
629
 
        # Make and consume sub generators, one per substream type:
630
 
        while self.first_bytes is not None:
631
 
            substream = NetworkRecordStream(self.iter_substream_bytes())
632
 
            # after substream is fully consumed, self.current_type is set to
633
 
            # the next type, and self.first_bytes is set to the matching bytes.
634
 
            yield self.current_type, wrap_and_count(pb, rc, substream)
635
 
        if rc:
636
 
            pb.update('Done', rc.max, rc.max)
637
 
        pb.finished()
638
 
 
639
 
    def seed_state(self):
640
 
        """Prepare the _ByteStreamDecoder to decode from the pack stream."""
641
 
        # Set a single generator we can use to get data from the pack stream.
642
 
        self.iter_pack_records = self.iter_stream_decoder()
643
 
        # Seed the very first subiterator with content; after this each one
644
 
        # seeds the next.
645
 
        list(self.iter_substream_bytes())
646
 
 
647
 
 
648
 
def _byte_stream_to_stream(byte_stream, record_counter=None):
649
 
    """Convert a byte stream into a format and a stream.
650
 
 
651
 
    :param byte_stream: A bytes iterator, as output by _stream_to_byte_stream.
652
 
    :return: (RepositoryFormat, stream_generator)
653
 
    """
654
 
    decoder = _ByteStreamDecoder(byte_stream, record_counter)
 
489
            substream = NetworkRecordStream([record_bytes])
 
490
            yield substream_type, substream.read()
 
491
        for bytes in byte_stream:
 
492
            stream_decoder.accept_bytes(bytes)
 
493
            for record in stream_decoder.read_pending_records():
 
494
                record_names, record_bytes = record
 
495
                record_name, = record_names
 
496
                substream_type = record_name[0]
 
497
                substream = NetworkRecordStream([record_bytes])
 
498
                yield substream_type, substream.read()
655
499
    for bytes in byte_stream:
656
 
        decoder.stream_decoder.accept_bytes(bytes)
657
 
        for record in decoder.stream_decoder.read_pending_records(max=1):
 
500
        stream_decoder.accept_bytes(bytes)
 
501
        for record in stream_decoder.read_pending_records(max=1):
658
502
            record_names, src_format_name = record
659
503
            src_format = network_format_registry.get(src_format_name)
660
 
            return src_format, decoder.record_stream()
 
504
            return src_format, record_stream()
661
505
 
662
506
 
663
507
class SmartServerRepositoryUnlock(SmartServerRepositoryRequest):
723
567
            temp.close()
724
568
 
725
569
    def _tarball_of_dir(self, dirname, compression, ofile):
726
 
        import tarfile
727
570
        filename = os.path.basename(ofile.name)
728
571
        tarball = tarfile.open(fileobj=ofile, name=filename,
729
572
            mode='w|' + compression)
807
650
            return SuccessfulSmartServerResponse(('ok', ))
808
651
 
809
652
 
810
 
class SmartServerRepositoryInsertStream_1_19(SmartServerRepositoryInsertStreamLocked):
811
 
    """Insert a record stream from a RemoteSink into a repository.
812
 
 
813
 
    Same as SmartServerRepositoryInsertStreamLocked, except:
814
 
     - the lock token argument is optional
815
 
     - servers that implement this verb accept 'inventory-delta' records in the
816
 
       stream.
817
 
 
818
 
    New in 1.19.
819
 
    """
820
 
 
821
 
    def do_repository_request(self, repository, resume_tokens, lock_token=None):
822
 
        """StreamSink.insert_stream for a remote repository."""
823
 
        SmartServerRepositoryInsertStreamLocked.do_repository_request(
824
 
            self, repository, resume_tokens, lock_token)
825
 
 
826
 
 
827
653
class SmartServerRepositoryInsertStream(SmartServerRepositoryInsertStreamLocked):
828
654
    """Insert a record stream from a RemoteSink into an unlocked repository.
829
655