1
# Copyright (C) 2006, 2007 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
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
17
"""Wire-level encoding and decoding of requests and responses for the smart
22
from cStringIO import StringIO
25
from bzrlib import debug
26
from bzrlib import errors
27
from bzrlib.smart import request
28
from bzrlib.trace import log_exception_quietly, mutter
31
# Protocol version strings. These are sent as prefixes of bzr requests and
32
# responses to identify the protocol version being used. (There are no version
33
# one strings because that version doesn't send any).
34
REQUEST_VERSION_TWO = 'bzr request 2\n'
35
RESPONSE_VERSION_TWO = 'bzr response 2\n'
38
def _recv_tuple(from_file):
39
req_line = from_file.readline()
40
return _decode_tuple(req_line)
43
def _decode_tuple(req_line):
44
if req_line == None or req_line == '':
46
if req_line[-1] != '\n':
47
raise errors.SmartProtocolError("request %r not terminated" % req_line)
48
return tuple(req_line[:-1].split('\x01'))
51
def _encode_tuple(args):
52
"""Encode the tuple args to a bytestream."""
53
return '\x01'.join(args) + '\n'
56
class SmartProtocolBase(object):
57
"""Methods common to client and server"""
59
# TODO: this only actually accomodates a single block; possibly should
60
# support multiple chunks?
61
def _encode_bulk_data(self, body):
62
"""Encode body as a bulk data chunk."""
63
return ''.join(('%d\n' % len(body), body, 'done\n'))
65
def _serialise_offsets(self, offsets):
66
"""Serialise a readv offset list."""
68
for start, length in offsets:
69
txt.append('%d,%d' % (start, length))
73
class SmartServerRequestProtocolOne(SmartProtocolBase):
74
"""Server-side encoding and decoding logic for smart version 1."""
76
def __init__(self, backing_transport, write_func):
77
self._backing_transport = backing_transport
78
self.excess_buffer = ''
79
self._finished = False
81
self.has_dispatched = False
83
self._body_decoder = None
84
self._write_func = write_func
86
def accept_bytes(self, bytes):
87
"""Take bytes, and advance the internal state machine appropriately.
89
:param bytes: must be a byte string
91
assert isinstance(bytes, str)
92
self.in_buffer += bytes
93
if not self.has_dispatched:
94
if '\n' not in self.in_buffer:
97
self.has_dispatched = True
99
first_line, self.in_buffer = self.in_buffer.split('\n', 1)
101
req_args = _decode_tuple(first_line)
102
self.request = request.SmartServerRequestHandler(
103
self._backing_transport, commands=request.request_handlers)
104
self.request.dispatch_command(req_args[0], req_args[1:])
105
if self.request.finished_reading:
107
self.excess_buffer = self.in_buffer
109
self._send_response(self.request.response)
110
except KeyboardInterrupt:
112
except Exception, exception:
113
# everything else: pass to client, flush, and quit
114
log_exception_quietly()
115
self._send_response(request.FailedSmartServerResponse(
116
('error', str(exception))))
119
if self.has_dispatched:
121
# nothing to do.XXX: this routine should be a single state
123
self.excess_buffer += self.in_buffer
126
if self._body_decoder is None:
127
self._body_decoder = LengthPrefixedBodyDecoder()
128
self._body_decoder.accept_bytes(self.in_buffer)
129
self.in_buffer = self._body_decoder.unused_data
130
body_data = self._body_decoder.read_pending_data()
131
self.request.accept_body(body_data)
132
if self._body_decoder.finished_reading:
133
self.request.end_of_body()
134
assert self.request.finished_reading, \
135
"no more body, request not finished"
136
if self.request.response is not None:
137
self._send_response(self.request.response)
138
self.excess_buffer = self.in_buffer
141
assert not self.request.finished_reading, \
142
"no response and we have finished reading."
144
def _send_response(self, response):
145
"""Send a smart server response down the output stream."""
146
assert not self._finished, 'response already sent'
149
self._finished = True
150
self._write_protocol_version()
151
self._write_success_or_failure_prefix(response)
152
self._write_func(_encode_tuple(args))
154
assert isinstance(body, str), 'body must be a str'
155
bytes = self._encode_bulk_data(body)
156
self._write_func(bytes)
158
def _write_protocol_version(self):
159
"""Write any prefixes this protocol requires.
161
Version one doesn't send protocol versions.
164
def _write_success_or_failure_prefix(self, response):
165
"""Write the protocol specific success/failure prefix.
167
For SmartServerRequestProtocolOne this is omitted but we
168
call is_successful to ensure that the response is valid.
170
response.is_successful()
172
def next_read_size(self):
175
if self._body_decoder is None:
178
return self._body_decoder.next_read_size()
181
class SmartServerRequestProtocolTwo(SmartServerRequestProtocolOne):
182
r"""Version two of the server side of the smart protocol.
184
This prefixes responses with the value of RESPONSE_VERSION_TWO.
187
def _write_success_or_failure_prefix(self, response):
188
"""Write the protocol specific success/failure prefix."""
189
if response.is_successful():
190
self._write_func('success\n')
192
self._write_func('failed\n')
194
def _write_protocol_version(self):
195
r"""Write any prefixes this protocol requires.
197
Version two sends the value of RESPONSE_VERSION_TWO.
199
self._write_func(RESPONSE_VERSION_TWO)
201
def _send_response(self, response):
202
"""Send a smart server response down the output stream."""
203
assert not self._finished, 'response already sent'
204
self._finished = True
205
self._write_protocol_version()
206
self._write_success_or_failure_prefix(response)
207
self._write_func(_encode_tuple(response.args))
208
if response.body is not None:
209
assert isinstance(response.body, str), 'body must be a str'
210
bytes = self._encode_bulk_data(response.body)
211
self._write_func(bytes)
212
elif response.body_stream is not None:
213
_send_stream(response.body_stream, self._write_func)
216
def _send_stream(stream, write_func):
217
_send_chunks(stream, write_func)
221
def _send_chunks(stream, write_func):
223
if isinstance(chunk, str):
224
bytes = "%x\n%s" % (len(chunk), chunk)
226
elif isinstance(chunk, request.FailedSmartServerResponse):
228
_send_chunks(chunk.args, write_func)
232
'Chunks must be str or FailedSmartServerResponse, got %r'
236
class _StatefulDecoder(object):
239
self.finished_reading = False
240
self.unused_data = ''
241
self.bytes_left = None
243
def accept_bytes(self, bytes):
244
"""Decode as much of bytes as possible.
246
If 'bytes' contains too much data it will be appended to
249
finished_reading will be set when no more data is required. Further
250
data will be appended to self.unused_data.
252
# accept_bytes is allowed to change the state
253
current_state = self.state_accept
254
self.state_accept(bytes)
255
while current_state != self.state_accept:
256
current_state = self.state_accept
257
self.state_accept('')
260
class ChunkedBodyDecoder(_StatefulDecoder):
261
"""Decoder for chunked body data.
263
This is very similar the HTTP's chunked encoding. See the description of
264
streamed body data in `doc/developers/network-protocol.txt` for details.
268
_StatefulDecoder.__init__(self)
269
self.state_accept = self._state_accept_expecting_length
271
self.chunk_in_progress = None
272
self.chunks = collections.deque()
274
self.error_in_progress = None
276
def next_read_size(self):
277
# Note: the shortest possible chunk is 2 bytes: '0\n', and the
278
# end-of-body marker is 4 bytes: 'END\n'.
279
if self.state_accept == self._state_accept_reading_chunk:
280
# We're expecting more chunk content. So we're expecting at least
281
# the rest of this chunk plus an END chunk.
282
return self.bytes_left + 4
283
elif self.state_accept == self._state_accept_expecting_length:
284
if self._in_buffer == '':
285
# We're expecting a chunk length. There's at least two bytes
286
# left: a digit plus '\n'.
289
# We're in the middle of reading a chunk length. So there's at
290
# least one byte left, the '\n' that terminates the length.
292
elif self.state_accept == self._state_accept_reading_unused:
295
raise AssertionError("Impossible state: %r" % (self.state_accept,))
297
def read_next_chunk(self):
299
return self.chunks.popleft()
303
def _extract_line(self):
304
pos = self._in_buffer.find('\n')
306
# We haven't read a complete length prefix yet, so there's nothing
309
line = self._in_buffer[:pos]
310
# Trim the prefix (including '\n' delimiter) from the _in_buffer.
311
self._in_buffer = self._in_buffer[pos+1:]
315
self.unused_data = self._in_buffer
316
self._in_buffer = None
317
self.state_accept = self._state_accept_reading_unused
319
error_args = tuple(self.error_in_progress)
320
self.chunks.append(request.FailedSmartServerResponse(error_args))
321
self.error_in_progress = None
322
self.finished_reading = True
324
def _state_accept_expecting_length(self, bytes):
325
self._in_buffer += bytes
326
prefix = self._extract_line()
328
# We haven't read a complete length prefix yet, so there's nothing
331
elif prefix == 'ERR':
333
self.error_in_progress = []
334
self._state_accept_expecting_length('')
336
elif prefix == 'END':
337
# We've read the end-of-body marker.
338
# Any further bytes are unused data, including the bytes left in
343
self.bytes_left = int(prefix, 16)
344
self.chunk_in_progress = ''
345
self.state_accept = self._state_accept_reading_chunk
347
def _state_accept_reading_chunk(self, bytes):
348
self._in_buffer += bytes
349
in_buffer_len = len(self._in_buffer)
350
self.chunk_in_progress += self._in_buffer[:self.bytes_left]
351
self._in_buffer = self._in_buffer[self.bytes_left:]
352
self.bytes_left -= in_buffer_len
353
if self.bytes_left <= 0:
354
# Finished with chunk
355
self.bytes_left = None
357
self.error_in_progress.append(self.chunk_in_progress)
359
self.chunks.append(self.chunk_in_progress)
360
self.chunk_in_progress = None
361
self.state_accept = self._state_accept_expecting_length
363
def _state_accept_reading_unused(self, bytes):
364
self.unused_data += bytes
367
class LengthPrefixedBodyDecoder(_StatefulDecoder):
368
"""Decodes the length-prefixed bulk data."""
371
_StatefulDecoder.__init__(self)
372
self.state_accept = self._state_accept_expecting_length
373
self.state_read = self._state_read_no_data
375
self._trailer_buffer = ''
377
def next_read_size(self):
378
if self.bytes_left is not None:
379
# Ideally we want to read all the remainder of the body and the
381
return self.bytes_left + 5
382
elif self.state_accept == self._state_accept_reading_trailer:
383
# Just the trailer left
384
return 5 - len(self._trailer_buffer)
385
elif self.state_accept == self._state_accept_expecting_length:
386
# There's still at least 6 bytes left ('\n' to end the length, plus
390
# Reading excess data. Either way, 1 byte at a time is fine.
393
def read_pending_data(self):
394
"""Return any pending data that has been decoded."""
395
return self.state_read()
397
def _state_accept_expecting_length(self, bytes):
398
self._in_buffer += bytes
399
pos = self._in_buffer.find('\n')
402
self.bytes_left = int(self._in_buffer[:pos])
403
self._in_buffer = self._in_buffer[pos+1:]
404
self.bytes_left -= len(self._in_buffer)
405
self.state_accept = self._state_accept_reading_body
406
self.state_read = self._state_read_in_buffer
408
def _state_accept_reading_body(self, bytes):
409
self._in_buffer += bytes
410
self.bytes_left -= len(bytes)
411
if self.bytes_left <= 0:
413
if self.bytes_left != 0:
414
self._trailer_buffer = self._in_buffer[self.bytes_left:]
415
self._in_buffer = self._in_buffer[:self.bytes_left]
416
self.bytes_left = None
417
self.state_accept = self._state_accept_reading_trailer
419
def _state_accept_reading_trailer(self, bytes):
420
self._trailer_buffer += bytes
421
# TODO: what if the trailer does not match "done\n"? Should this raise
422
# a ProtocolViolation exception?
423
if self._trailer_buffer.startswith('done\n'):
424
self.unused_data = self._trailer_buffer[len('done\n'):]
425
self.state_accept = self._state_accept_reading_unused
426
self.finished_reading = True
428
def _state_accept_reading_unused(self, bytes):
429
self.unused_data += bytes
431
def _state_read_no_data(self):
434
def _state_read_in_buffer(self):
435
result = self._in_buffer
440
class SmartClientRequestProtocolOne(SmartProtocolBase):
441
"""The client-side protocol for smart version 1."""
443
def __init__(self, request):
444
"""Construct a SmartClientRequestProtocolOne.
446
:param request: A SmartClientMediumRequest to serialise onto and
449
self._request = request
450
self._body_buffer = None
451
self._request_start_time = None
453
def call(self, *args):
454
if 'hpss' in debug.debug_flags:
455
mutter('hpss call: %s', repr(args)[1:-1])
456
self._request_start_time = time.time()
457
self._write_args(args)
458
self._request.finished_writing()
460
def call_with_body_bytes(self, args, body):
461
"""Make a remote call of args with body bytes 'body'.
463
After calling this, call read_response_tuple to find the result out.
465
if 'hpss' in debug.debug_flags:
466
mutter('hpss call w/body: %s (%r...)', repr(args)[1:-1], body[:20])
467
mutter(' %d bytes', len(body))
468
self._request_start_time = time.time()
469
self._write_args(args)
470
bytes = self._encode_bulk_data(body)
471
self._request.accept_bytes(bytes)
472
self._request.finished_writing()
474
def call_with_body_readv_array(self, args, body):
475
"""Make a remote call with a readv array.
477
The body is encoded with one line per readv offset pair. The numbers in
478
each pair are separated by a comma, and no trailing \n is emitted.
480
if 'hpss' in debug.debug_flags:
481
mutter('hpss call w/readv: %s', repr(args)[1:-1])
482
self._request_start_time = time.time()
483
self._write_args(args)
484
readv_bytes = self._serialise_offsets(body)
485
bytes = self._encode_bulk_data(readv_bytes)
486
self._request.accept_bytes(bytes)
487
self._request.finished_writing()
488
if 'hpss' in debug.debug_flags:
489
mutter(' %d bytes in readv request', len(readv_bytes))
491
def cancel_read_body(self):
492
"""After expecting a body, a response code may indicate one otherwise.
494
This method lets the domain client inform the protocol that no body
495
will be transmitted. This is a terminal method: after calling it the
496
protocol is not able to be used further.
498
self._request.finished_reading()
500
def read_response_tuple(self, expect_body=False):
501
"""Read a response tuple from the wire.
503
This should only be called once.
505
result = self._recv_tuple()
506
if 'hpss' in debug.debug_flags:
507
if self._request_start_time is not None:
508
mutter(' result: %6.3fs %s',
509
time.time() - self._request_start_time,
511
self._request_start_time = None
513
mutter(' result: %s', repr(result)[1:-1])
515
self._request.finished_reading()
518
def read_body_bytes(self, count=-1):
519
"""Read bytes from the body, decoding into a byte stream.
521
We read all bytes at once to ensure we've checked the trailer for
522
errors, and then feed the buffer back as read_body_bytes is called.
524
if self._body_buffer is not None:
525
return self._body_buffer.read(count)
526
_body_decoder = LengthPrefixedBodyDecoder()
528
while not _body_decoder.finished_reading:
529
bytes_wanted = _body_decoder.next_read_size()
530
bytes = self._request.read_bytes(bytes_wanted)
531
_body_decoder.accept_bytes(bytes)
532
self._request.finished_reading()
533
self._body_buffer = StringIO(_body_decoder.read_pending_data())
534
# XXX: TODO check the trailer result.
535
if 'hpss' in debug.debug_flags:
536
mutter(' %d body bytes read',
537
len(self._body_buffer.getvalue()))
538
return self._body_buffer.read(count)
540
def _recv_tuple(self):
541
"""Receive a tuple from the medium request."""
542
return _decode_tuple(self._recv_line())
544
def _recv_line(self):
545
"""Read an entire line from the medium request."""
547
while not line or line[-1] != '\n':
548
# TODO: this is inefficient - but tuples are short.
549
new_char = self._request.read_bytes(1)
551
assert new_char != '', "end of file reading from server."
554
def query_version(self):
555
"""Return protocol version number of the server."""
557
resp = self.read_response_tuple()
558
if resp == ('ok', '1'):
560
elif resp == ('ok', '2'):
563
raise errors.SmartProtocolError("bad response %r" % (resp,))
565
def _write_args(self, args):
566
self._write_protocol_version()
567
bytes = _encode_tuple(args)
568
self._request.accept_bytes(bytes)
570
def _write_protocol_version(self):
571
"""Write any prefixes this protocol requires.
573
Version one doesn't send protocol versions.
577
class SmartClientRequestProtocolTwo(SmartClientRequestProtocolOne):
578
"""Version two of the client side of the smart protocol.
580
This prefixes the request with the value of REQUEST_VERSION_TWO.
583
def read_response_tuple(self, expect_body=False):
584
"""Read a response tuple from the wire.
586
This should only be called once.
588
version = self._request.read_line()
589
if version != RESPONSE_VERSION_TWO:
590
raise errors.SmartProtocolError('bad protocol marker %r' % version)
591
response_status = self._recv_line()
592
if response_status not in ('success\n', 'failed\n'):
593
raise errors.SmartProtocolError(
594
'bad protocol status %r' % response_status)
595
self.response_status = response_status == 'success\n'
596
return SmartClientRequestProtocolOne.read_response_tuple(self, expect_body)
598
def _write_protocol_version(self):
599
"""Write any prefixes this protocol requires.
601
Version two sends the value of REQUEST_VERSION_TWO.
603
self._request.accept_bytes(REQUEST_VERSION_TWO)
605
def read_streamed_body(self):
606
"""Read bytes from the body, decoding into a byte stream.
608
_body_decoder = ChunkedBodyDecoder()
609
while not _body_decoder.finished_reading:
610
bytes_wanted = _body_decoder.next_read_size()
611
bytes = self._request.read_bytes(bytes_wanted)
612
_body_decoder.accept_bytes(bytes)
613
for body_bytes in iter(_body_decoder.read_next_chunk, None):
614
if 'hpss' in debug.debug_flags:
615
mutter(' %d byte chunk read',
618
self._request.finished_reading()
added
removed
bzrlib/smart/protocol.py
