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
28
from bzrlib import debug
29
from bzrlib import errors
30
from bzrlib.smart import message, request
31
from bzrlib.trace import log_exception_quietly, mutter
32
from bzrlib.util.bencode import bdecode, bencode
35
# Protocol version strings. These are sent as prefixes of bzr requests and
36
# responses to identify the protocol version being used. (There are no version
37
# one strings because that version doesn't send any).
38
REQUEST_VERSION_TWO = 'bzr request 2\n'
39
RESPONSE_VERSION_TWO = 'bzr response 2\n'
41
MESSAGE_VERSION_THREE = 'bzr message 3 (bzr 1.6)\n'
42
RESPONSE_VERSION_THREE = REQUEST_VERSION_THREE = MESSAGE_VERSION_THREE
45
def _recv_tuple(from_file):
46
req_line = from_file.readline()
47
return _decode_tuple(req_line)
50
def _decode_tuple(req_line):
51
if req_line is None or req_line == '':
53
if req_line[-1] != '\n':
54
raise errors.SmartProtocolError("request %r not terminated" % req_line)
55
return tuple(req_line[:-1].split('\x01'))
58
def _encode_tuple(args):
59
"""Encode the tuple args to a bytestream."""
60
return '\x01'.join(args) + '\n'
63
class Requester(object):
64
"""Abstract base class for an object that can issue requests on a smart
68
def call(self, *args):
69
"""Make a remote call.
71
:param args: the arguments of this call.
73
raise NotImplementedError(self.call)
75
def call_with_body_bytes(self, args, body):
76
"""Make a remote call with a body.
78
:param args: the arguments of this call.
80
:param body: the body to send with the request.
82
raise NotImplementedError(self.call_with_body_bytes)
84
def call_with_body_readv_array(self, args, body):
85
"""Make a remote call with a readv array.
87
:param args: the arguments of this call.
88
:type body: iterable of (start, length) tuples.
89
:param body: the readv ranges to send with this request.
91
raise NotImplementedError(self.call_with_body_readv_array)
93
def set_headers(self, headers):
94
raise NotImplementedError(self.set_headers)
97
class SmartProtocolBase(object):
98
"""Methods common to client and server"""
100
# TODO: this only actually accomodates a single block; possibly should
101
# support multiple chunks?
102
def _encode_bulk_data(self, body):
103
"""Encode body as a bulk data chunk."""
104
return ''.join(('%d\n' % len(body), body, 'done\n'))
106
def _serialise_offsets(self, offsets):
107
"""Serialise a readv offset list."""
109
for start, length in offsets:
110
txt.append('%d,%d' % (start, length))
111
return '\n'.join(txt)
114
class SmartServerRequestProtocolOne(SmartProtocolBase):
115
"""Server-side encoding and decoding logic for smart version 1."""
117
def __init__(self, backing_transport, write_func, root_client_path='/'):
118
self._backing_transport = backing_transport
119
self._root_client_path = root_client_path
120
self.unused_data = ''
121
self._finished = False
123
self._has_dispatched = False
125
self._body_decoder = None
126
self._write_func = write_func
128
def accept_bytes(self, bytes):
129
"""Take bytes, and advance the internal state machine appropriately.
131
:param bytes: must be a byte string
133
if not isinstance(bytes, str):
134
raise ValueError(bytes)
135
self.in_buffer += bytes
136
if not self._has_dispatched:
137
if '\n' not in self.in_buffer:
138
# no command line yet
140
self._has_dispatched = True
142
first_line, self.in_buffer = self.in_buffer.split('\n', 1)
144
req_args = _decode_tuple(first_line)
145
self.request = request.SmartServerRequestHandler(
146
self._backing_transport, commands=request.request_handlers,
147
root_client_path=self._root_client_path)
148
self.request.dispatch_command(req_args[0], req_args[1:])
149
if self.request.finished_reading:
151
self.unused_data = self.in_buffer
153
self._send_response(self.request.response)
154
except KeyboardInterrupt:
156
except errors.UnknownSmartMethod, err:
157
protocol_error = errors.SmartProtocolError(
158
"bad request %r" % (err.verb,))
159
failure = request.FailedSmartServerResponse(
160
('error', str(protocol_error)))
161
self._send_response(failure)
163
except Exception, exception:
164
# everything else: pass to client, flush, and quit
165
log_exception_quietly()
166
self._send_response(request.FailedSmartServerResponse(
167
('error', str(exception))))
170
if self._has_dispatched:
172
# nothing to do.XXX: this routine should be a single state
174
self.unused_data += self.in_buffer
177
if self._body_decoder is None:
178
self._body_decoder = LengthPrefixedBodyDecoder()
179
self._body_decoder.accept_bytes(self.in_buffer)
180
self.in_buffer = self._body_decoder.unused_data
181
body_data = self._body_decoder.read_pending_data()
182
self.request.accept_body(body_data)
183
if self._body_decoder.finished_reading:
184
self.request.end_of_body()
185
if not self.request.finished_reading:
186
raise AssertionError("no more body, request not finished")
187
if self.request.response is not None:
188
self._send_response(self.request.response)
189
self.unused_data = self.in_buffer
192
if self.request.finished_reading:
193
raise AssertionError(
194
"no response and we have finished reading.")
196
def _send_response(self, response):
197
"""Send a smart server response down the output stream."""
199
raise AssertionError('response already sent')
202
self._finished = True
203
self._write_protocol_version()
204
self._write_success_or_failure_prefix(response)
205
self._write_func(_encode_tuple(args))
207
if not isinstance(body, str):
208
raise ValueError(body)
209
bytes = self._encode_bulk_data(body)
210
self._write_func(bytes)
212
def _write_protocol_version(self):
213
"""Write any prefixes this protocol requires.
215
Version one doesn't send protocol versions.
218
def _write_success_or_failure_prefix(self, response):
219
"""Write the protocol specific success/failure prefix.
221
For SmartServerRequestProtocolOne this is omitted but we
222
call is_successful to ensure that the response is valid.
224
response.is_successful()
226
def next_read_size(self):
229
if self._body_decoder is None:
232
return self._body_decoder.next_read_size()
235
class SmartServerRequestProtocolTwo(SmartServerRequestProtocolOne):
236
r"""Version two of the server side of the smart protocol.
238
This prefixes responses with the value of RESPONSE_VERSION_TWO.
241
response_marker = RESPONSE_VERSION_TWO
242
request_marker = REQUEST_VERSION_TWO
244
def _write_success_or_failure_prefix(self, response):
245
"""Write the protocol specific success/failure prefix."""
246
if response.is_successful():
247
self._write_func('success\n')
249
self._write_func('failed\n')
251
def _write_protocol_version(self):
252
r"""Write any prefixes this protocol requires.
254
Version two sends the value of RESPONSE_VERSION_TWO.
256
self._write_func(self.response_marker)
258
def _send_response(self, response):
259
"""Send a smart server response down the output stream."""
261
raise AssertionError('response already sent')
262
self._finished = True
263
self._write_protocol_version()
264
self._write_success_or_failure_prefix(response)
265
self._write_func(_encode_tuple(response.args))
266
if response.body is not None:
267
if not isinstance(response.body, str):
268
raise AssertionError('body must be a str')
269
if not (response.body_stream is None):
270
raise AssertionError(
271
'body_stream and body cannot both be set')
272
bytes = self._encode_bulk_data(response.body)
273
self._write_func(bytes)
274
elif response.body_stream is not None:
275
_send_stream(response.body_stream, self._write_func)
278
def _send_stream(stream, write_func):
279
write_func('chunked\n')
280
_send_chunks(stream, write_func)
284
def _send_chunks(stream, write_func):
286
if isinstance(chunk, str):
287
bytes = "%x\n%s" % (len(chunk), chunk)
289
elif isinstance(chunk, request.FailedSmartServerResponse):
291
_send_chunks(chunk.args, write_func)
294
raise errors.BzrError(
295
'Chunks must be str or FailedSmartServerResponse, got %r'
299
class _NeedMoreBytes(Exception):
300
"""Raise this inside a _StatefulDecoder to stop decoding until more bytes
304
def __init__(self, count=None):
307
:param count: the total number of bytes needed by the current state.
308
May be None if the number of bytes needed is unknown.
313
class _StatefulDecoder(object):
314
"""Base class for writing state machines to decode byte streams.
316
Subclasses should provide a self.state_accept attribute that accepts bytes
317
and, if appropriate, updates self.state_accept to a different function.
318
accept_bytes will call state_accept as often as necessary to make sure the
319
state machine has progressed as far as possible before it returns.
321
See ProtocolThreeDecoder for an example subclass.
325
self.finished_reading = False
326
self._in_buffer_list = []
327
self._in_buffer_len = 0
328
self.unused_data = ''
329
self.bytes_left = None
330
self._number_needed_bytes = None
332
def _get_in_buffer(self):
333
if len(self._in_buffer_list) == 1:
334
return self._in_buffer_list[0]
335
in_buffer = ''.join(self._in_buffer_list)
336
if len(in_buffer) != self._in_buffer_len:
337
raise AssertionError(
338
"Length of buffer did not match expected value: %s != %s"
339
% self._in_buffer_len, len(in_buffer))
340
self._in_buffer_list = [in_buffer]
343
def _get_in_bytes(self, count):
344
"""Grab X bytes from the input_buffer.
346
Callers should have already checked that self._in_buffer_len is >
347
count. Note, this does not consume the bytes from the buffer. The
348
caller will still need to call _get_in_buffer() and then
349
_set_in_buffer() if they actually need to consume the bytes.
351
# check if we can yield the bytes from just the first entry in our list
352
if len(self._in_buffer_list) == 0:
353
raise AssertionError('Callers must be sure we have buffered bytes'
354
' before calling _get_in_bytes')
355
if len(self._in_buffer_list[0]) > count:
356
return self._in_buffer_list[0][:count]
357
# We can't yield it from the first buffer, so collapse all buffers, and
359
in_buf = self._get_in_buffer()
360
return in_buf[:count]
362
def _set_in_buffer(self, new_buf):
363
if new_buf is not None:
364
self._in_buffer_list = [new_buf]
365
self._in_buffer_len = len(new_buf)
367
self._in_buffer_list = []
368
self._in_buffer_len = 0
370
def accept_bytes(self, bytes):
371
"""Decode as much of bytes as possible.
373
If 'bytes' contains too much data it will be appended to
376
finished_reading will be set when no more data is required. Further
377
data will be appended to self.unused_data.
379
# accept_bytes is allowed to change the state
380
self._number_needed_bytes = None
381
# lsprof puts a very large amount of time on this specific call for
383
self._in_buffer_list.append(bytes)
384
self._in_buffer_len += len(bytes)
386
# Run the function for the current state.
387
current_state = self.state_accept
389
while current_state != self.state_accept:
390
# The current state has changed. Run the function for the new
391
# current state, so that it can:
392
# - decode any unconsumed bytes left in a buffer, and
393
# - signal how many more bytes are expected (via raising
395
current_state = self.state_accept
397
except _NeedMoreBytes, e:
398
self._number_needed_bytes = e.count
401
class ChunkedBodyDecoder(_StatefulDecoder):
402
"""Decoder for chunked body data.
404
This is very similar the HTTP's chunked encoding. See the description of
405
streamed body data in `doc/developers/network-protocol.txt` for details.
409
_StatefulDecoder.__init__(self)
410
self.state_accept = self._state_accept_expecting_header
411
self.chunk_in_progress = None
412
self.chunks = collections.deque()
414
self.error_in_progress = None
416
def next_read_size(self):
417
# Note: the shortest possible chunk is 2 bytes: '0\n', and the
418
# end-of-body marker is 4 bytes: 'END\n'.
419
if self.state_accept == self._state_accept_reading_chunk:
420
# We're expecting more chunk content. So we're expecting at least
421
# the rest of this chunk plus an END chunk.
422
return self.bytes_left + 4
423
elif self.state_accept == self._state_accept_expecting_length:
424
if self._in_buffer_len == 0:
425
# We're expecting a chunk length. There's at least two bytes
426
# left: a digit plus '\n'.
429
# We're in the middle of reading a chunk length. So there's at
430
# least one byte left, the '\n' that terminates the length.
432
elif self.state_accept == self._state_accept_reading_unused:
434
elif self.state_accept == self._state_accept_expecting_header:
435
return max(0, len('chunked\n') - self._in_buffer_len)
437
raise AssertionError("Impossible state: %r" % (self.state_accept,))
439
def read_next_chunk(self):
441
return self.chunks.popleft()
445
def _extract_line(self):
446
in_buf = self._get_in_buffer()
447
pos = in_buf.find('\n')
449
# We haven't read a complete line yet, so request more bytes before
451
raise _NeedMoreBytes(1)
453
# Trim the prefix (including '\n' delimiter) from the _in_buffer.
454
self._set_in_buffer(in_buf[pos+1:])
458
self.unused_data = self._get_in_buffer()
459
# self._in_buffer = None
460
self._in_buffer_list = []
461
self._in_buffer_len = 0
462
self.state_accept = self._state_accept_reading_unused
464
error_args = tuple(self.error_in_progress)
465
self.chunks.append(request.FailedSmartServerResponse(error_args))
466
self.error_in_progress = None
467
self.finished_reading = True
469
def _state_accept_expecting_header(self):
470
prefix = self._extract_line()
471
if prefix == 'chunked':
472
self.state_accept = self._state_accept_expecting_length
474
raise errors.SmartProtocolError(
475
'Bad chunked body header: "%s"' % (prefix,))
477
def _state_accept_expecting_length(self):
478
prefix = self._extract_line()
481
self.error_in_progress = []
482
self._state_accept_expecting_length()
484
elif prefix == 'END':
485
# We've read the end-of-body marker.
486
# Any further bytes are unused data, including the bytes left in
491
self.bytes_left = int(prefix, 16)
492
self.chunk_in_progress = ''
493
self.state_accept = self._state_accept_reading_chunk
495
def _state_accept_reading_chunk(self):
496
in_buf = self._get_in_buffer()
497
in_buffer_len = len(in_buf)
498
self.chunk_in_progress += in_buf[:self.bytes_left]
499
self._set_in_buffer(in_buf[self.bytes_left:])
500
self.bytes_left -= in_buffer_len
501
if self.bytes_left <= 0:
502
# Finished with chunk
503
self.bytes_left = None
505
self.error_in_progress.append(self.chunk_in_progress)
507
self.chunks.append(self.chunk_in_progress)
508
self.chunk_in_progress = None
509
self.state_accept = self._state_accept_expecting_length
511
def _state_accept_reading_unused(self):
512
self.unused_data += self._get_in_buffer()
513
self._in_buffer_list = []
516
class LengthPrefixedBodyDecoder(_StatefulDecoder):
517
"""Decodes the length-prefixed bulk data."""
520
_StatefulDecoder.__init__(self)
521
self.state_accept = self._state_accept_expecting_length
522
self.state_read = self._state_read_no_data
524
self._trailer_buffer = ''
526
def next_read_size(self):
527
if self.bytes_left is not None:
528
# Ideally we want to read all the remainder of the body and the
530
return self.bytes_left + 5
531
elif self.state_accept == self._state_accept_reading_trailer:
532
# Just the trailer left
533
return 5 - len(self._trailer_buffer)
534
elif self.state_accept == self._state_accept_expecting_length:
535
# There's still at least 6 bytes left ('\n' to end the length, plus
539
# Reading excess data. Either way, 1 byte at a time is fine.
542
def read_pending_data(self):
543
"""Return any pending data that has been decoded."""
544
return self.state_read()
546
def _state_accept_expecting_length(self):
547
in_buf = self._get_in_buffer()
548
pos = in_buf.find('\n')
551
self.bytes_left = int(in_buf[:pos])
552
self._set_in_buffer(in_buf[pos+1:])
553
self.state_accept = self._state_accept_reading_body
554
self.state_read = self._state_read_body_buffer
556
def _state_accept_reading_body(self):
557
in_buf = self._get_in_buffer()
559
self.bytes_left -= len(in_buf)
560
self._set_in_buffer(None)
561
if self.bytes_left <= 0:
563
if self.bytes_left != 0:
564
self._trailer_buffer = self._body[self.bytes_left:]
565
self._body = self._body[:self.bytes_left]
566
self.bytes_left = None
567
self.state_accept = self._state_accept_reading_trailer
569
def _state_accept_reading_trailer(self):
570
self._trailer_buffer += self._get_in_buffer()
571
self._set_in_buffer(None)
572
# TODO: what if the trailer does not match "done\n"? Should this raise
573
# a ProtocolViolation exception?
574
if self._trailer_buffer.startswith('done\n'):
575
self.unused_data = self._trailer_buffer[len('done\n'):]
576
self.state_accept = self._state_accept_reading_unused
577
self.finished_reading = True
579
def _state_accept_reading_unused(self):
580
self.unused_data += self._get_in_buffer()
581
self._set_in_buffer(None)
583
def _state_read_no_data(self):
586
def _state_read_body_buffer(self):
592
class SmartClientRequestProtocolOne(SmartProtocolBase, Requester,
593
message.ResponseHandler):
594
"""The client-side protocol for smart version 1."""
596
def __init__(self, request):
597
"""Construct a SmartClientRequestProtocolOne.
599
:param request: A SmartClientMediumRequest to serialise onto and
602
self._request = request
603
self._body_buffer = None
604
self._request_start_time = None
605
self._last_verb = None
608
def set_headers(self, headers):
609
self._headers = dict(headers)
611
def call(self, *args):
612
if 'hpss' in debug.debug_flags:
613
mutter('hpss call: %s', repr(args)[1:-1])
614
if getattr(self._request._medium, 'base', None) is not None:
615
mutter(' (to %s)', self._request._medium.base)
616
self._request_start_time = time.time()
617
self._write_args(args)
618
self._request.finished_writing()
619
self._last_verb = args[0]
621
def call_with_body_bytes(self, args, body):
622
"""Make a remote call of args with body bytes 'body'.
624
After calling this, call read_response_tuple to find the result out.
626
if 'hpss' in debug.debug_flags:
627
mutter('hpss call w/body: %s (%r...)', repr(args)[1:-1], body[:20])
628
if getattr(self._request._medium, '_path', None) is not None:
629
mutter(' (to %s)', self._request._medium._path)
630
mutter(' %d bytes', len(body))
631
self._request_start_time = time.time()
632
if 'hpssdetail' in debug.debug_flags:
633
mutter('hpss body content: %s', body)
634
self._write_args(args)
635
bytes = self._encode_bulk_data(body)
636
self._request.accept_bytes(bytes)
637
self._request.finished_writing()
638
self._last_verb = args[0]
640
def call_with_body_readv_array(self, args, body):
641
"""Make a remote call with a readv array.
643
The body is encoded with one line per readv offset pair. The numbers in
644
each pair are separated by a comma, and no trailing \n is emitted.
646
if 'hpss' in debug.debug_flags:
647
mutter('hpss call w/readv: %s', repr(args)[1:-1])
648
if getattr(self._request._medium, '_path', None) is not None:
649
mutter(' (to %s)', self._request._medium._path)
650
self._request_start_time = time.time()
651
self._write_args(args)
652
readv_bytes = self._serialise_offsets(body)
653
bytes = self._encode_bulk_data(readv_bytes)
654
self._request.accept_bytes(bytes)
655
self._request.finished_writing()
656
if 'hpss' in debug.debug_flags:
657
mutter(' %d bytes in readv request', len(readv_bytes))
658
self._last_verb = args[0]
660
def cancel_read_body(self):
661
"""After expecting a body, a response code may indicate one otherwise.
663
This method lets the domain client inform the protocol that no body
664
will be transmitted. This is a terminal method: after calling it the
665
protocol is not able to be used further.
667
self._request.finished_reading()
669
def _read_response_tuple(self):
670
result = self._recv_tuple()
671
if 'hpss' in debug.debug_flags:
672
if self._request_start_time is not None:
673
mutter(' result: %6.3fs %s',
674
time.time() - self._request_start_time,
676
self._request_start_time = None
678
mutter(' result: %s', repr(result)[1:-1])
681
def read_response_tuple(self, expect_body=False):
682
"""Read a response tuple from the wire.
684
This should only be called once.
686
result = self._read_response_tuple()
687
self._response_is_unknown_method(result)
688
self._raise_args_if_error(result)
690
self._request.finished_reading()
693
def _raise_args_if_error(self, result_tuple):
694
# Later protocol versions have an explicit flag in the protocol to say
695
# if an error response is "failed" or not. In version 1 we don't have
696
# that luxury. So here is a complete list of errors that can be
697
# returned in response to existing version 1 smart requests. Responses
698
# starting with these codes are always "failed" responses.
705
'UnicodeEncodeError',
706
'UnicodeDecodeError',
712
'UnlockableTransport',
718
if result_tuple[0] in v1_error_codes:
719
self._request.finished_reading()
720
raise errors.ErrorFromSmartServer(result_tuple)
722
def _response_is_unknown_method(self, result_tuple):
723
"""Raise UnexpectedSmartServerResponse if the response is an 'unknonwn
724
method' response to the request.
726
:param response: The response from a smart client call_expecting_body
728
:param verb: The verb used in that call.
729
:raises: UnexpectedSmartServerResponse
731
if (result_tuple == ('error', "Generic bzr smart protocol error: "
732
"bad request '%s'" % self._last_verb) or
733
result_tuple == ('error', "Generic bzr smart protocol error: "
734
"bad request u'%s'" % self._last_verb)):
735
# The response will have no body, so we've finished reading.
736
self._request.finished_reading()
737
raise errors.UnknownSmartMethod(self._last_verb)
739
def read_body_bytes(self, count=-1):
740
"""Read bytes from the body, decoding into a byte stream.
742
We read all bytes at once to ensure we've checked the trailer for
743
errors, and then feed the buffer back as read_body_bytes is called.
745
if self._body_buffer is not None:
746
return self._body_buffer.read(count)
747
_body_decoder = LengthPrefixedBodyDecoder()
749
while not _body_decoder.finished_reading:
750
bytes = self._request.read_bytes(_body_decoder.next_read_size())
752
# end of file encountered reading from server
753
raise errors.ConnectionReset(
754
"Connection lost while reading response body.")
755
_body_decoder.accept_bytes(bytes)
756
self._request.finished_reading()
757
self._body_buffer = StringIO(_body_decoder.read_pending_data())
758
# XXX: TODO check the trailer result.
759
if 'hpss' in debug.debug_flags:
760
mutter(' %d body bytes read',
761
len(self._body_buffer.getvalue()))
762
return self._body_buffer.read(count)
764
def _recv_tuple(self):
765
"""Receive a tuple from the medium request."""
766
return _decode_tuple(self._request.read_line())
768
def query_version(self):
769
"""Return protocol version number of the server."""
771
resp = self.read_response_tuple()
772
if resp == ('ok', '1'):
774
elif resp == ('ok', '2'):
777
raise errors.SmartProtocolError("bad response %r" % (resp,))
779
def _write_args(self, args):
780
self._write_protocol_version()
781
bytes = _encode_tuple(args)
782
self._request.accept_bytes(bytes)
784
def _write_protocol_version(self):
785
"""Write any prefixes this protocol requires.
787
Version one doesn't send protocol versions.
791
class SmartClientRequestProtocolTwo(SmartClientRequestProtocolOne):
792
"""Version two of the client side of the smart protocol.
794
This prefixes the request with the value of REQUEST_VERSION_TWO.
797
response_marker = RESPONSE_VERSION_TWO
798
request_marker = REQUEST_VERSION_TWO
800
def read_response_tuple(self, expect_body=False):
801
"""Read a response tuple from the wire.
803
This should only be called once.
805
version = self._request.read_line()
806
if version != self.response_marker:
807
self._request.finished_reading()
808
raise errors.UnexpectedProtocolVersionMarker(version)
809
response_status = self._request.read_line()
810
result = SmartClientRequestProtocolOne._read_response_tuple(self)
811
self._response_is_unknown_method(result)
812
if response_status == 'success\n':
813
self.response_status = True
815
self._request.finished_reading()
817
elif response_status == 'failed\n':
818
self.response_status = False
819
self._request.finished_reading()
820
raise errors.ErrorFromSmartServer(result)
822
raise errors.SmartProtocolError(
823
'bad protocol status %r' % response_status)
825
def _write_protocol_version(self):
826
"""Write any prefixes this protocol requires.
828
Version two sends the value of REQUEST_VERSION_TWO.
830
self._request.accept_bytes(self.request_marker)
832
def read_streamed_body(self):
833
"""Read bytes from the body, decoding into a byte stream.
835
# Read no more than 64k at a time so that we don't risk error 10055 (no
836
# buffer space available) on Windows.
837
_body_decoder = ChunkedBodyDecoder()
838
while not _body_decoder.finished_reading:
839
bytes = self._request.read_bytes(_body_decoder.next_read_size())
841
# end of file encountered reading from server
842
raise errors.ConnectionReset(
843
"Connection lost while reading streamed body.")
844
_body_decoder.accept_bytes(bytes)
845
for body_bytes in iter(_body_decoder.read_next_chunk, None):
846
if 'hpss' in debug.debug_flags and type(body_bytes) is str:
847
mutter(' %d byte chunk read',
850
self._request.finished_reading()
853
def build_server_protocol_three(backing_transport, write_func,
855
request_handler = request.SmartServerRequestHandler(
856
backing_transport, commands=request.request_handlers,
857
root_client_path=root_client_path)
858
responder = ProtocolThreeResponder(write_func)
859
message_handler = message.ConventionalRequestHandler(request_handler, responder)
860
return ProtocolThreeDecoder(message_handler)
863
class ProtocolThreeDecoder(_StatefulDecoder):
865
response_marker = RESPONSE_VERSION_THREE
866
request_marker = REQUEST_VERSION_THREE
868
def __init__(self, message_handler, expect_version_marker=False):
869
_StatefulDecoder.__init__(self)
870
self._has_dispatched = False
872
if expect_version_marker:
873
self.state_accept = self._state_accept_expecting_protocol_version
874
# We're expecting at least the protocol version marker + some
876
self._number_needed_bytes = len(MESSAGE_VERSION_THREE) + 4
878
self.state_accept = self._state_accept_expecting_headers
879
self._number_needed_bytes = 4
880
self.decoding_failed = False
881
self.request_handler = self.message_handler = message_handler
883
def accept_bytes(self, bytes):
884
self._number_needed_bytes = None
886
_StatefulDecoder.accept_bytes(self, bytes)
887
except KeyboardInterrupt:
889
except errors.SmartMessageHandlerError, exception:
890
# We do *not* set self.decoding_failed here. The message handler
891
# has raised an error, but the decoder is still able to parse bytes
892
# and determine when this message ends.
893
log_exception_quietly()
894
self.message_handler.protocol_error(exception.exc_value)
895
# The state machine is ready to continue decoding, but the
896
# exception has interrupted the loop that runs the state machine.
897
# So we call accept_bytes again to restart it.
898
self.accept_bytes('')
899
except Exception, exception:
900
# The decoder itself has raised an exception. We cannot continue
902
self.decoding_failed = True
903
if isinstance(exception, errors.UnexpectedProtocolVersionMarker):
904
# This happens during normal operation when the client tries a
905
# protocol version the server doesn't understand, so no need to
906
# log a traceback every time.
907
# Note that this can only happen when
908
# expect_version_marker=True, which is only the case on the
912
log_exception_quietly()
913
self.message_handler.protocol_error(exception)
915
def _extract_length_prefixed_bytes(self):
916
if self._in_buffer_len < 4:
917
# A length prefix by itself is 4 bytes, and we don't even have that
919
raise _NeedMoreBytes(4)
920
(length,) = struct.unpack('!L', self._get_in_bytes(4))
921
end_of_bytes = 4 + length
922
if self._in_buffer_len < end_of_bytes:
923
# We haven't yet read as many bytes as the length-prefix says there
925
raise _NeedMoreBytes(end_of_bytes)
926
# Extract the bytes from the buffer.
927
in_buf = self._get_in_buffer()
928
bytes = in_buf[4:end_of_bytes]
929
self._set_in_buffer(in_buf[end_of_bytes:])
932
def _extract_prefixed_bencoded_data(self):
933
prefixed_bytes = self._extract_length_prefixed_bytes()
935
decoded = bdecode(prefixed_bytes)
937
raise errors.SmartProtocolError(
938
'Bytes %r not bencoded' % (prefixed_bytes,))
941
def _extract_single_byte(self):
942
if self._in_buffer_len == 0:
943
# The buffer is empty
944
raise _NeedMoreBytes(1)
945
in_buf = self._get_in_buffer()
947
self._set_in_buffer(in_buf[1:])
950
def _state_accept_expecting_protocol_version(self):
951
needed_bytes = len(MESSAGE_VERSION_THREE) - self._in_buffer_len
952
in_buf = self._get_in_buffer()
954
# We don't have enough bytes to check if the protocol version
955
# marker is right. But we can check if it is already wrong by
956
# checking that the start of MESSAGE_VERSION_THREE matches what
958
# [In fact, if the remote end isn't bzr we might never receive
959
# len(MESSAGE_VERSION_THREE) bytes. So if the bytes we have so far
960
# are wrong then we should just raise immediately rather than
962
if not MESSAGE_VERSION_THREE.startswith(in_buf):
963
# We have enough bytes to know the protocol version is wrong
964
raise errors.UnexpectedProtocolVersionMarker(in_buf)
965
raise _NeedMoreBytes(len(MESSAGE_VERSION_THREE))
966
if not in_buf.startswith(MESSAGE_VERSION_THREE):
967
raise errors.UnexpectedProtocolVersionMarker(in_buf)
968
self._set_in_buffer(in_buf[len(MESSAGE_VERSION_THREE):])
969
self.state_accept = self._state_accept_expecting_headers
971
def _state_accept_expecting_headers(self):
972
decoded = self._extract_prefixed_bencoded_data()
973
if type(decoded) is not dict:
974
raise errors.SmartProtocolError(
975
'Header object %r is not a dict' % (decoded,))
976
self.state_accept = self._state_accept_expecting_message_part
978
self.message_handler.headers_received(decoded)
980
raise errors.SmartMessageHandlerError(sys.exc_info())
982
def _state_accept_expecting_message_part(self):
983
message_part_kind = self._extract_single_byte()
984
if message_part_kind == 'o':
985
self.state_accept = self._state_accept_expecting_one_byte
986
elif message_part_kind == 's':
987
self.state_accept = self._state_accept_expecting_structure
988
elif message_part_kind == 'b':
989
self.state_accept = self._state_accept_expecting_bytes
990
elif message_part_kind == 'e':
993
raise errors.SmartProtocolError(
994
'Bad message kind byte: %r' % (message_part_kind,))
996
def _state_accept_expecting_one_byte(self):
997
byte = self._extract_single_byte()
998
self.state_accept = self._state_accept_expecting_message_part
1000
self.message_handler.byte_part_received(byte)
1002
raise errors.SmartMessageHandlerError(sys.exc_info())
1004
def _state_accept_expecting_bytes(self):
1005
# XXX: this should not buffer whole message part, but instead deliver
1006
# the bytes as they arrive.
1007
prefixed_bytes = self._extract_length_prefixed_bytes()
1008
self.state_accept = self._state_accept_expecting_message_part
1010
self.message_handler.bytes_part_received(prefixed_bytes)
1012
raise errors.SmartMessageHandlerError(sys.exc_info())
1014
def _state_accept_expecting_structure(self):
1015
structure = self._extract_prefixed_bencoded_data()
1016
self.state_accept = self._state_accept_expecting_message_part
1018
self.message_handler.structure_part_received(structure)
1020
raise errors.SmartMessageHandlerError(sys.exc_info())
1023
self.unused_data = self._get_in_buffer()
1024
self._set_in_buffer(None)
1025
self.state_accept = self._state_accept_reading_unused
1027
self.message_handler.end_received()
1029
raise errors.SmartMessageHandlerError(sys.exc_info())
1031
def _state_accept_reading_unused(self):
1032
self.unused_data = self._get_in_buffer()
1033
self._set_in_buffer(None)
1035
def next_read_size(self):
1036
if self.state_accept == self._state_accept_reading_unused:
1038
elif self.decoding_failed:
1039
# An exception occured while processing this message, probably from
1040
# self.message_handler. We're not sure that this state machine is
1041
# in a consistent state, so just signal that we're done (i.e. give
1045
if self._number_needed_bytes is not None:
1046
return self._number_needed_bytes - self._in_buffer_len
1048
raise AssertionError("don't know how many bytes are expected!")
1051
class _ProtocolThreeEncoder(object):
1053
response_marker = request_marker = MESSAGE_VERSION_THREE
1055
def __init__(self, write_func):
1057
self._real_write_func = write_func
1059
def _write_func(self, bytes):
1064
self._real_write_func(self._buf)
1067
def _serialise_offsets(self, offsets):
1068
"""Serialise a readv offset list."""
1070
for start, length in offsets:
1071
txt.append('%d,%d' % (start, length))
1072
return '\n'.join(txt)
1074
def _write_protocol_version(self):
1075
self._write_func(MESSAGE_VERSION_THREE)
1077
def _write_prefixed_bencode(self, structure):
1078
bytes = bencode(structure)
1079
self._write_func(struct.pack('!L', len(bytes)))
1080
self._write_func(bytes)
1082
def _write_headers(self, headers):
1083
self._write_prefixed_bencode(headers)
1085
def _write_structure(self, args):
1086
self._write_func('s')
1089
if type(arg) is unicode:
1090
utf8_args.append(arg.encode('utf8'))
1092
utf8_args.append(arg)
1093
self._write_prefixed_bencode(utf8_args)
1095
def _write_end(self):
1096
self._write_func('e')
1099
def _write_prefixed_body(self, bytes):
1100
self._write_func('b')
1101
self._write_func(struct.pack('!L', len(bytes)))
1102
self._write_func(bytes)
1104
def _write_error_status(self):
1105
self._write_func('oE')
1107
def _write_success_status(self):
1108
self._write_func('oS')
1111
class ProtocolThreeResponder(_ProtocolThreeEncoder):
1113
def __init__(self, write_func):
1114
_ProtocolThreeEncoder.__init__(self, write_func)
1115
self.response_sent = False
1116
self._headers = {'Software version': bzrlib.__version__}
1118
def send_error(self, exception):
1119
if self.response_sent:
1120
raise AssertionError(
1121
"send_error(%s) called, but response already sent."
1123
if isinstance(exception, errors.UnknownSmartMethod):
1124
failure = request.FailedSmartServerResponse(
1125
('UnknownMethod', exception.verb))
1126
self.send_response(failure)
1128
self.response_sent = True
1129
self._write_protocol_version()
1130
self._write_headers(self._headers)
1131
self._write_error_status()
1132
self._write_structure(('error', str(exception)))
1135
def send_response(self, response):
1136
if self.response_sent:
1137
raise AssertionError(
1138
"send_response(%r) called, but response already sent."
1140
self.response_sent = True
1141
self._write_protocol_version()
1142
self._write_headers(self._headers)
1143
if response.is_successful():
1144
self._write_success_status()
1146
self._write_error_status()
1147
self._write_structure(response.args)
1148
if response.body is not None:
1149
self._write_prefixed_body(response.body)
1150
elif response.body_stream is not None:
1151
for chunk in response.body_stream:
1152
self._write_prefixed_body(chunk)
1157
class ProtocolThreeRequester(_ProtocolThreeEncoder, Requester):
1159
def __init__(self, medium_request):
1160
_ProtocolThreeEncoder.__init__(self, medium_request.accept_bytes)
1161
self._medium_request = medium_request
1164
def set_headers(self, headers):
1165
self._headers = headers.copy()
1167
def call(self, *args):
1168
if 'hpss' in debug.debug_flags:
1169
mutter('hpss call: %s', repr(args)[1:-1])
1170
base = getattr(self._medium_request._medium, 'base', None)
1171
if base is not None:
1172
mutter(' (to %s)', base)
1173
self._request_start_time = time.time()
1174
self._write_protocol_version()
1175
self._write_headers(self._headers)
1176
self._write_structure(args)
1178
self._medium_request.finished_writing()
1180
def call_with_body_bytes(self, args, body):
1181
"""Make a remote call of args with body bytes 'body'.
1183
After calling this, call read_response_tuple to find the result out.
1185
if 'hpss' in debug.debug_flags:
1186
mutter('hpss call w/body: %s (%r...)', repr(args)[1:-1], body[:20])
1187
path = getattr(self._medium_request._medium, '_path', None)
1188
if path is not None:
1189
mutter(' (to %s)', path)
1190
mutter(' %d bytes', len(body))
1191
self._request_start_time = time.time()
1192
self._write_protocol_version()
1193
self._write_headers(self._headers)
1194
self._write_structure(args)
1195
self._write_prefixed_body(body)
1197
self._medium_request.finished_writing()
1199
def call_with_body_readv_array(self, args, body):
1200
"""Make a remote call with a readv array.
1202
The body is encoded with one line per readv offset pair. The numbers in
1203
each pair are separated by a comma, and no trailing \n is emitted.
1205
if 'hpss' in debug.debug_flags:
1206
mutter('hpss call w/readv: %s', repr(args)[1:-1])
1207
path = getattr(self._medium_request._medium, '_path', None)
1208
if path is not None:
1209
mutter(' (to %s)', path)
1210
self._request_start_time = time.time()
1211
self._write_protocol_version()
1212
self._write_headers(self._headers)
1213
self._write_structure(args)
1214
readv_bytes = self._serialise_offsets(body)
1215
if 'hpss' in debug.debug_flags:
1216
mutter(' %d bytes in readv request', len(readv_bytes))
1217
self._write_prefixed_body(readv_bytes)
1219
self._medium_request.finished_writing()
added
removed
bzrlib/smart/protocol.py
