1
# Copyright (C) 2007 Canonical Ltd
1
# Copyright (C) 2007, 2009, 2010 Canonical Ltd
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
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
17
"""Container format for Bazaar data.
48
50
def _check_name_encoding(name):
49
51
"""Check that 'name' is valid UTF-8.
51
53
This is separate from _check_name because UTF-8 decoding is relatively
52
54
expensive, and we usually want to avoid it.
62
64
class ContainerSerialiser(object):
63
65
"""A helper class for serialising containers.
65
67
It simply returns bytes from method calls to 'begin', 'end' and
66
68
'bytes_record'. You may find ContainerWriter to be a more convenient
75
77
"""Return the bytes to finish a container."""
78
def bytes_record(self, bytes, names):
79
"""Return the bytes for a Bytes record with the given name and
80
def bytes_header(self, length, names):
81
"""Return the header for a Bytes record."""
83
83
byte_sections = ["B"]
85
byte_sections.append(str(len(bytes)) + "\n")
85
byte_sections.append(str(length) + "\n")
87
87
for name_tuple in names:
88
88
# Make sure we're writing valid names. Note that we will leave a
92
92
byte_sections.append('\x00'.join(name_tuple) + "\n")
94
94
byte_sections.append("\n")
95
# Finally, the contents.
96
byte_sections.append(bytes)
97
# XXX: This causes a memory copy of bytes in size, but is usually
98
# faster than two write calls (12 vs 13 seconds to output a gig of
99
# 1k records.) - results may differ on significantly larger records
100
# like .iso's but as they should be rare in any case and thus not
101
# likely to be the common case. The biggest issue is causing extreme
102
# memory pressure in that case. One possibly improvement here is to
103
# check the size of the content before deciding to join here vs call
105
95
return ''.join(byte_sections)
97
def bytes_record(self, bytes, names):
98
"""Return the bytes for a Bytes record with the given name and
101
If the content may be large, construct the header separately and then
102
stream out the contents.
104
return self.bytes_header(len(bytes), names) + bytes
108
107
class ContainerWriter(object):
109
108
"""A class for writing containers to a file.
113
112
introduced by the begin() and end() methods.
115
# Join up headers with the body if writing fewer than this many bytes:
116
# trades off memory usage and copying to do less IO ops.
117
_JOIN_WRITES_THRESHOLD = 100000
116
119
def __init__(self, write_func):
139
142
def add_bytes_record(self, bytes, names):
140
143
"""Add a Bytes record with the given names.
142
145
:param bytes: The bytes to insert.
143
146
:param names: The names to give the inserted bytes. Each name is
144
147
a tuple of bytestrings. The bytestrings may not contain
151
154
and thus are only suitable for use by a ContainerReader.
153
156
current_offset = self.current_offset
154
serialised_record = self._serialiser.bytes_record(bytes, names)
155
self.write_func(serialised_record)
158
if length < self._JOIN_WRITES_THRESHOLD:
159
self.write_func(self._serialiser.bytes_header(length, names)
162
self.write_func(self._serialiser.bytes_header(length, names))
163
self.write_func(bytes)
156
164
self.records_written += 1
157
165
# return a memo of where we wrote data to allow random access.
158
166
return current_offset, self.current_offset - current_offset
161
169
class ReadVFile(object):
162
"""Adapt a readv result iterator to a file like protocol."""
170
"""Adapt a readv result iterator to a file like protocol.
172
The readv result must support the iterator protocol returning (offset,
176
# XXX: This could be a generic transport class, as other code may want to
177
# gradually consume the readv result.
164
179
def __init__(self, readv_result):
180
"""Construct a new ReadVFile wrapper.
182
:seealso: make_readv_reader
184
:param readv_result: the most recent readv result - list or generator
186
# readv can return a sequence or an iterator, but we require an
187
# iterator to know how much has been consumed.
188
readv_result = iter(readv_result)
165
189
self.readv_result = readv_result
166
# the most recent readv result block
167
190
self._string = None
170
193
if (self._string is None or
171
194
self._string.tell() == self._string_length):
172
length, data = self.readv_result.next()
195
offset, data = self.readv_result.next()
173
196
self._string_length = len(data)
174
197
self._string = StringIO(data)
178
201
result = self._string.read(length)
179
202
if len(result) < length:
180
raise errors.BzrError('request for too much data from a readv hunk.')
203
raise errors.BzrError('wanted %d bytes but next '
204
'hunk only contains %d: %r...' %
205
(length, len(result), result[:20]))
183
208
def readline(self):
186
211
result = self._string.readline()
187
212
if self._string.tell() == self._string_length and result[-1] != '\n':
188
raise errors.BzrError('short readline in the readvfile hunk.')
213
raise errors.BzrError('short readline in the readvfile hunk: %r'
234
260
is a ``list`` and bytes is a function that takes one argument,
237
You **must not** call the callable after advancing the interator to the
263
You **must not** call the callable after advancing the iterator to the
238
264
next record. That is, this code is invalid::
240
266
record_iter = container.iter_records()
241
267
names1, callable1 = record_iter.next()
242
268
names2, callable2 = record_iter.next()
243
269
bytes1 = callable1(None)
245
271
As it will give incorrect results and invalidate the state of the
248
:raises ContainerError: if any sort of containter corruption is
274
:raises ContainerError: if any sort of container corruption is
249
275
detected, e.g. UnknownContainerFormatError is the format of the
250
276
container is unrecognised.
251
277
:seealso: ContainerReader.read
253
279
self._read_format()
254
280
return self._iter_records()
256
282
def iter_record_objects(self):
257
283
"""Iterate over the container, yielding each record as it is read.
260
286
methods. Like with iter_records, it is not safe to use a record object
261
287
after advancing the iterator to yield next record.
263
:raises ContainerError: if any sort of containter corruption is
289
:raises ContainerError: if any sort of container corruption is
264
290
detected, e.g. UnknownContainerFormatError is the format of the
265
291
container is unrecognised.
266
292
:seealso: iter_records
268
294
self._read_format()
269
295
return self._iter_record_objects()
271
297
def _iter_records(self):
272
298
for record in self._iter_record_objects():
273
299
yield record.read()
315
341
# risk that the same unicode string has been encoded two
316
342
# different ways.
317
343
if name_tuple in all_names:
318
raise errors.DuplicateRecordNameError(name_tuple)
344
raise errors.DuplicateRecordNameError(name_tuple[0])
319
345
all_names.add(name_tuple)
320
346
excess_bytes = self.reader_func(1)
321
347
if excess_bytes != '':
407
433
last_buffer_length = None
408
434
cur_buffer_length = len(self._buffer)
409
while cur_buffer_length != last_buffer_length:
435
last_state_handler = None
436
while (cur_buffer_length != last_buffer_length
437
or last_state_handler != self._state_handler):
410
438
last_buffer_length = cur_buffer_length
439
last_state_handler = self._state_handler
411
440
self._state_handler()
412
441
cur_buffer_length = len(self._buffer)
414
def read_pending_records(self):
415
records = self._parsed_records
416
self._parsed_records = []
443
def read_pending_records(self, max=None):
445
records = self._parsed_records[:max]
446
del self._parsed_records[:max]
449
records = self._parsed_records
450
self._parsed_records = []
419
453
def _consume_line(self):
420
454
"""Take a line out of the buffer, and return the line.
468
502
for name_part in name_parts:
469
503
_check_name(name_part)
470
504
self._current_record_names.append(name_parts)
472
506
def _state_expecting_body(self):
473
507
if len(self._buffer) >= self._current_record_length:
474
508
body_bytes = self._buffer[:self._current_record_length]