« back to all changes in this revision
Viewing changes to bzrlib/fetch.py
- browse files at revision 4060.1.3
- compare with another revision
- download diff
- download tarball
- view history from revision 4060.1.3
- Committer: Robert Collins
- Date: 2009-02-27 13:05:36 UTC
- mto: This revision was merged to the branch mainline in revision 4067.
- Revision ID: robertc@robertcollins.net-20090227130536-wsqoyhyt3n11nc8d
Implement the separate source component for fetch - repository.StreamSource.
added
removed
bzrlib/fetch.py
67
67
if set, try to limit to the data this revision references.
68
68
69
69
after running:
70
count_copied -- number of revisions copied
71
70
72
71
This should not be used directly, it's essential a object to encapsulate
73
72
the logic in InterRepository.fetch().
83
82
like to remove this parameter.
84
83
"""
85
84
# result variables.
85
self.count_copied = 0
86
86
self.failed_revisions = []
87
self.count_copied = 0
88
87
if to_repository.has_same_location(from_repository):
89
88
# repository.fetch should be taking care of this case.
90
89
raise errors.BzrError('RepoFetcher run '
146
145
# item_keys_introduced_by should have a richer API than it does at the
147
146
# moment, so that it can feed the progress information back to this
148
147
# function?
148
if (self.from_repository._format.rich_root_data and
149
not self.to_repository._format.rich_root_data):
150
raise errors.IncompatibleRepositories(
151
self.from_repository, self.to_repository,
152
"different rich-root support")
149
153
self.pb = bzrlib.ui.ui_factory.nested_progress_bar()
150
154
try:
155
source = self.from_repository._get_source(
156
self.to_repository._format)
157
stream = source.get_stream(search)
151
158
from_format = self.from_repository._format
152
stream = self.get_stream(search, pp)
153
159
resume_tokens, missing_keys = self.sink.insert_stream(
154
160
stream, from_format, [])
155
161
if missing_keys:
156
stream = self.get_stream_for_missing_keys(missing_keys)
162
stream = source.get_stream_for_missing_keys(missing_keys)
157
163
resume_tokens, missing_keys = self.sink.insert_stream(
158
164
stream, from_format, resume_tokens)
159
165
if missing_keys:
165
171
"second push failed to commit the fetch %r." % (
166
172
resume_tokens,))
167
173
self.sink.finished()
174
self.count_copied = source.count_copied
168
175
finally:
169
176
if self.pb is not None:
170
177
self.pb.finished()
171
178
172
def get_stream(self, search, pp):
173
phase = 'file'
174
revs = search.get_keys()
175
graph = self.from_repository.get_graph()
176
revs = list(graph.iter_topo_order(revs))
177
data_to_fetch = self.from_repository.item_keys_introduced_by(
178
revs, self.pb)
179
text_keys = []
180
for knit_kind, file_id, revisions in data_to_fetch:
181
if knit_kind != phase:
182
phase = knit_kind
183
# Make a new progress bar for this phase
184
self.pb.finished()
185
pp.next_phase()
186
self.pb = bzrlib.ui.ui_factory.nested_progress_bar()
187
if knit_kind == "file":
188
# Accumulate file texts
189
text_keys.extend([(file_id, revision) for revision in
190
revisions])
191
elif knit_kind == "inventory":
192
# Now copy the file texts.
193
from_texts = self.from_repository.texts
194
yield ('texts', from_texts.get_record_stream(
195
text_keys, self.to_repository._format._fetch_order,
196
not self.to_repository._format._fetch_uses_deltas))
197
# Cause an error if a text occurs after we have done the
198
# copy.
199
text_keys = None
200
# Before we process the inventory we generate the root
201
# texts (if necessary) so that the inventories references
202
# will be valid.
203
for _ in self._generate_root_texts(revs):
204
yield _
205
# NB: This currently reopens the inventory weave in source;
206
# using a single stream interface instead would avoid this.
207
self.pb.update("fetch inventory", 0, 1)
208
from_weave = self.from_repository.inventories
209
# we fetch only the referenced inventories because we do not
210
# know for unselected inventories whether all their required
211
# texts are present in the other repository - it could be
212
# corrupt.
213
yield ('inventories', from_weave.get_record_stream(
214
[(rev_id,) for rev_id in revs],
215
self.inventory_fetch_order(),
216
not self.delta_on_metadata()))
217
elif knit_kind == "signatures":
218
# Nothing to do here; this will be taken care of when
219
# _fetch_revision_texts happens.
220
pass
221
elif knit_kind == "revisions":
222
for _ in self._fetch_revision_texts(revs, self.pb):
223
yield _
224
else:
225
raise AssertionError("Unknown knit kind %r" % knit_kind)
226
self.count_copied += len(revs)
227
228
def get_stream_for_missing_keys(self, missing_keys):
229
# missing keys can only occur when we are byte copying and not
230
# translating (because translation means we don't send
231
# unreconstructable deltas ever).
232
keys = {}
233
keys['texts'] = set()
234
keys['revisions'] = set()
235
keys['inventories'] = set()
236
keys['signatures'] = set()
237
for key in missing_keys:
238
keys[key[0]].add(key[1:])
239
if len(keys['revisions']):
240
# If we allowed copying revisions at this point, we could end up
241
# copying a revision without copying its required texts: a
242
# violation of the requirements for repository integrity.
243
raise AssertionError(
244
'cannot copy revisions to fill in missing deltas %s' % (
245
keys['revisions'],))
246
for substream_kind, keys in keys.iteritems():
247
vf = getattr(self.from_repository, substream_kind)
248
# Ask for full texts always so that we don't need more round trips
249
# after this stream.
250
stream = vf.get_record_stream(keys,
251
self.to_repository._format._fetch_order, True)
252
yield substream_kind, stream
253
254
179
def _revids_to_fetch(self):
255
180
"""Determines the exact revisions needed from self.from_repository to
256
181
install self._last_revision in self.to_repository.
271
196
except errors.NoSuchRevision, e:
272
197
raise InstallFailed([self._last_revision])
273
198
274
def _fetch_revision_texts(self, revs, pb):
275
# fetch signatures first and then the revision texts
276
# may need to be a InterRevisionStore call here.
277
from_sf = self.from_repository.signatures
278
# A missing signature is just skipped.
279
keys = [(rev_id,) for rev_id in revs]
280
signatures = filter_absent(from_sf.get_record_stream(
281
keys,
282
self.to_repository._format._fetch_order,
283
not self.to_repository._format._fetch_uses_deltas))
284
# If a revision has a delta, this is actually expanded inside the
285
# insert_record_stream code now, which is an alternate fix for
286
# bug #261339
287
from_rf = self.from_repository.revisions
288
revisions = from_rf.get_record_stream(
289
keys,
290
self.to_repository._format._fetch_order,
291
not self.delta_on_metadata())
292
return [('signatures', signatures), ('revisions', revisions)]
293
294
def _generate_root_texts(self, revs):
295
"""This will be called by __fetch between fetching weave texts and
296
fetching the inventory weave.
297
298
Subclasses should override this if they need to generate root texts
299
after fetching weave texts.
300
"""
301
return []
302
303
def inventory_fetch_order(self):
304
return self.to_repository._format._fetch_order
305
306
def delta_on_metadata(self):
307
src_serializer = self.from_repository._format._serializer
308
target_serializer = self.to_repository._format._serializer
309
return (self.to_repository._format._fetch_uses_deltas and
310
src_serializer == target_serializer)
311
312
199
313
200
class Inter1and2Helper(object):
314
201
"""Helper for operations that convert data from model 1 and 2
397
284
rev_id_to_root_id.get(parent, root_id) == root_id)
398
285
yield FulltextContentFactory(key, parent_keys, None, '')
399
286
return [('texts', yield_roots())]
400
401
402
class Model1toKnit2Fetcher(RepoFetcher):
403
"""Fetch from a Model1 repository into a Knit2 repository
404
"""
405
def __init__(self, to_repository, from_repository, last_revision=None,
406
pb=None, find_ghosts=True):
407
self.helper = Inter1and2Helper(from_repository)
408
RepoFetcher.__init__(self, to_repository, from_repository,
409
last_revision, pb, find_ghosts)
410
411
def _generate_root_texts(self, revs):
412
return self.helper.generate_root_texts(revs)
413
414
def inventory_fetch_order(self):
415
return 'topological'
416
417
Knit1to2Fetcher = Model1toKnit2Fetcher
Loggerhead is a web-based interface for Breezy
Version: 2.0.1