1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
# Copyright (C) 2006 by Canonical Ltd
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as published by
# the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""VersionedFile based revision store.
This stores revisions as individual entries in a knit, and signatures in a
parallel knit.
"""
import bzrlib
import bzrlib.errors as errors
from bzrlib.knit import KnitVersionedFile, KnitPlainFactory
from bzrlib.store.revision import RevisionStore
from bzrlib.store.versioned import VersionedFileStore
from bzrlib.transport import get_transport
class KnitRevisionStoreFactory(object):
"""Factory to create a KnitRevisionStore for testing."""
def create(self, url):
"""Create a revision store at url."""
t = get_transport(url)
t.mkdir('revision-store')
versioned_file_store = VersionedFileStore(
t.clone('revision-store'),
precious=True,
versionedfile_class=KnitVersionedFile,
versionedfile_kwargs={'delta':False, 'factory':KnitPlainFactory()})
return KnitRevisionStore(versioned_file_store)
def __str__(self):
return "KnitRevisionStore"
class KnitRevisionStore(RevisionStore):
"""A RevisionStore layering on a VersionedFileStore."""
def __init__(self, versioned_file_store):
"""Create a KnitRevisionStore object.
:param versioned_file_store: the text store to use for storing
revisions and signatures.
"""
super(KnitRevisionStore, self).__init__()
self.versioned_file_store = versioned_file_store
def _add_revision(self, revision, revision_as_file, transaction):
"""Template method helper to store revision in this store."""
# FIXME: make this ghost aware at the knit level
rf = self.get_revision_file(transaction)
self.get_revision_file(transaction).add_lines_with_ghosts(
revision.revision_id,
revision.parent_ids,
bzrlib.osutils.split_lines(revision_as_file.read()))
def add_revision_signature_text(self, revision_id, signature_text, transaction):
"""See RevisionStore.add_revision_signature_text()."""
self.get_signature_file(transaction).add_lines(
revision_id, [], bzrlib.osutils.split_lines(signature_text))
def all_revision_ids(self, transaction):
"""See RevisionStore.all_revision_ids()."""
rev_file = self.get_revision_file(transaction)
return rev_file.get_ancestry(rev_file.versions())
def get_revisions(self, revision_ids, transaction):
"""See RevisionStore.get_revisions()."""
texts = self._get_serialized_revisions(revision_ids, transaction)
revisions = []
try:
for text, revision_id in zip(texts, revision_ids):
r = self._serializer.read_revision_from_string(text)
assert r.revision_id == revision_id
revisions.append(r)
except SyntaxError, e:
raise errors.BzrError('failed to unpack revision_xml',
[revision_id,
str(e)])
return revisions
def _get_serialized_revisions(self, revision_ids, transaction):
texts = []
vf = self.get_revision_file(transaction)
try:
return vf.get_texts(revision_ids)
except (errors.RevisionNotPresent), e:
raise errors.NoSuchRevision(self, e.revision_id)
def _get_revision_xml(self, revision_id, transaction):
try:
return self.get_revision_file(transaction).get_text(revision_id)
except (errors.RevisionNotPresent):
raise errors.NoSuchRevision(self, revision_id)
def get_revision_file(self, transaction):
"""Get the revision versioned file object."""
return self.versioned_file_store.get_weave_or_empty('revisions', transaction)
def get_signature_file(self, transaction):
"""Get the signature text versioned file object."""
return self.versioned_file_store.get_weave_or_empty('signatures', transaction)
def _get_signature_text(self, revision_id, transaction):
"""See RevisionStore._get_signature_text()."""
try:
return self.get_signature_file(transaction).get_text(revision_id)
except errors.RevisionNotPresent:
raise errors.NoSuchRevision(self, revision_id)
def has_revision_id(self, revision_id, transaction):
"""True if the store contains revision_id."""
return (revision_id is None
or self.get_revision_file(transaction).has_version(revision_id))
def _has_signature(self, revision_id, transaction):
"""See RevisionStore._has_signature()."""
return self.get_signature_file(transaction).has_version(revision_id)
def total_size(self, transaction):
""" See RevisionStore.total_size()."""
return (len(self.all_revision_ids(transaction)),
self.versioned_file_store.total_size()[1])
|