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
327
self.unused_data = ''
328
self.bytes_left = None
329
self._number_needed_bytes = None
331
def accept_bytes(self, bytes):
332
"""Decode as much of bytes as possible.
334
If 'bytes' contains too much data it will be appended to
337
finished_reading will be set when no more data is required. Further
338
data will be appended to self.unused_data.
340
# accept_bytes is allowed to change the state
341
current_state = self.state_accept
342
self._number_needed_bytes = None
343
self._in_buffer += bytes
345
# Run the function for the current state.
347
while current_state != self.state_accept:
348
# The current state has changed. Run the function for the new
349
# current state, so that it can:
350
# - decode any unconsumed bytes left in a buffer, and
351
# - signal how many more bytes are expected (via raising
353
current_state = self.state_accept
355
except _NeedMoreBytes, e:
356
self._number_needed_bytes = e.count
359
class ChunkedBodyDecoder(_StatefulDecoder):
360
"""Decoder for chunked body data.
362
This is very similar the HTTP's chunked encoding. See the description of
363
streamed body data in `doc/developers/network-protocol.txt` for details.
367
_StatefulDecoder.__init__(self)
368
self.state_accept = self._state_accept_expecting_header
369
self.chunk_in_progress = None
370
self.chunks = collections.deque()
372
self.error_in_progress = None
374
def next_read_size(self):
375
# Note: the shortest possible chunk is 2 bytes: '0\n', and the
376
# end-of-body marker is 4 bytes: 'END\n'.
377
if self.state_accept == self._state_accept_reading_chunk:
378
# We're expecting more chunk content. So we're expecting at least
379
# the rest of this chunk plus an END chunk.
380
return self.bytes_left + 4
381
elif self.state_accept == self._state_accept_expecting_length:
382
if self._in_buffer == '':
383
# We're expecting a chunk length. There's at least two bytes
384
# left: a digit plus '\n'.
387
# We're in the middle of reading a chunk length. So there's at
388
# least one byte left, the '\n' that terminates the length.
390
elif self.state_accept == self._state_accept_reading_unused:
392
elif self.state_accept == self._state_accept_expecting_header:
393
return max(0, len('chunked\n') - len(self._in_buffer))
395
raise AssertionError("Impossible state: %r" % (self.state_accept,))
397
def read_next_chunk(self):
399
return self.chunks.popleft()
403
def _extract_line(self):
404
pos = self._in_buffer.find('\n')
406
# We haven't read a complete line yet, so request more bytes before
408
raise _NeedMoreBytes(1)
409
line = self._in_buffer[:pos]
410
# Trim the prefix (including '\n' delimiter) from the _in_buffer.
411
self._in_buffer = self._in_buffer[pos+1:]
415
self.unused_data = self._in_buffer
417
self.state_accept = self._state_accept_reading_unused
419
error_args = tuple(self.error_in_progress)
420
self.chunks.append(request.FailedSmartServerResponse(error_args))
421
self.error_in_progress = None
422
self.finished_reading = True
424
def _state_accept_expecting_header(self):
425
prefix = self._extract_line()
426
if prefix == 'chunked':
427
self.state_accept = self._state_accept_expecting_length
429
raise errors.SmartProtocolError(
430
'Bad chunked body header: "%s"' % (prefix,))
432
def _state_accept_expecting_length(self):
433
prefix = self._extract_line()
436
self.error_in_progress = []
437
self._state_accept_expecting_length()
439
elif prefix == 'END':
440
# We've read the end-of-body marker.
441
# Any further bytes are unused data, including the bytes left in
446
self.bytes_left = int(prefix, 16)
447
self.chunk_in_progress = ''
448
self.state_accept = self._state_accept_reading_chunk
450
def _state_accept_reading_chunk(self):
451
in_buffer_len = len(self._in_buffer)
452
self.chunk_in_progress += self._in_buffer[:self.bytes_left]
453
self._in_buffer = self._in_buffer[self.bytes_left:]
454
self.bytes_left -= in_buffer_len
455
if self.bytes_left <= 0:
456
# Finished with chunk
457
self.bytes_left = None
459
self.error_in_progress.append(self.chunk_in_progress)
461
self.chunks.append(self.chunk_in_progress)
462
self.chunk_in_progress = None
463
self.state_accept = self._state_accept_expecting_length
465
def _state_accept_reading_unused(self):
466
self.unused_data += self._in_buffer
470
class LengthPrefixedBodyDecoder(_StatefulDecoder):
471
"""Decodes the length-prefixed bulk data."""
474
_StatefulDecoder.__init__(self)
475
self.state_accept = self._state_accept_expecting_length
476
self.state_read = self._state_read_no_data
478
self._trailer_buffer = ''
480
def next_read_size(self):
481
if self.bytes_left is not None:
482
# Ideally we want to read all the remainder of the body and the
484
return self.bytes_left + 5
485
elif self.state_accept == self._state_accept_reading_trailer:
486
# Just the trailer left
487
return 5 - len(self._trailer_buffer)
488
elif self.state_accept == self._state_accept_expecting_length:
489
# There's still at least 6 bytes left ('\n' to end the length, plus
493
# Reading excess data. Either way, 1 byte at a time is fine.
496
def read_pending_data(self):
497
"""Return any pending data that has been decoded."""
498
return self.state_read()
500
def _state_accept_expecting_length(self):
501
pos = self._in_buffer.find('\n')
504
self.bytes_left = int(self._in_buffer[:pos])
505
self._in_buffer = self._in_buffer[pos+1:]
506
self.state_accept = self._state_accept_reading_body
507
self.state_read = self._state_read_body_buffer
509
def _state_accept_reading_body(self):
510
self._body += self._in_buffer
511
self.bytes_left -= len(self._in_buffer)
513
if self.bytes_left <= 0:
515
if self.bytes_left != 0:
516
self._trailer_buffer = self._body[self.bytes_left:]
517
self._body = self._body[:self.bytes_left]
518
self.bytes_left = None
519
self.state_accept = self._state_accept_reading_trailer
521
def _state_accept_reading_trailer(self):
522
self._trailer_buffer += self._in_buffer
524
# TODO: what if the trailer does not match "done\n"? Should this raise
525
# a ProtocolViolation exception?
526
if self._trailer_buffer.startswith('done\n'):
527
self.unused_data = self._trailer_buffer[len('done\n'):]
528
self.state_accept = self._state_accept_reading_unused
529
self.finished_reading = True
531
def _state_accept_reading_unused(self):
532
self.unused_data += self._in_buffer
535
def _state_read_no_data(self):
538
def _state_read_body_buffer(self):
544
class SmartClientRequestProtocolOne(SmartProtocolBase, Requester,
545
message.ResponseHandler):
546
"""The client-side protocol for smart version 1."""
548
def __init__(self, request):
549
"""Construct a SmartClientRequestProtocolOne.
551
:param request: A SmartClientMediumRequest to serialise onto and
554
self._request = request
555
self._body_buffer = None
556
self._request_start_time = None
557
self._last_verb = None
560
def set_headers(self, headers):
561
self._headers = dict(headers)
563
def call(self, *args):
564
if 'hpss' in debug.debug_flags:
565
mutter('hpss call: %s', repr(args)[1:-1])
566
if getattr(self._request._medium, 'base', None) is not None:
567
mutter(' (to %s)', self._request._medium.base)
568
self._request_start_time = time.time()
569
self._write_args(args)
570
self._request.finished_writing()
571
self._last_verb = args[0]
573
def call_with_body_bytes(self, args, body):
574
"""Make a remote call of args with body bytes 'body'.
576
After calling this, call read_response_tuple to find the result out.
578
if 'hpss' in debug.debug_flags:
579
mutter('hpss call w/body: %s (%r...)', repr(args)[1:-1], body[:20])
580
if getattr(self._request._medium, '_path', None) is not None:
581
mutter(' (to %s)', self._request._medium._path)
582
mutter(' %d bytes', len(body))
583
self._request_start_time = time.time()
584
if 'hpssdetail' in debug.debug_flags:
585
mutter('hpss body content: %s', body)
586
self._write_args(args)
587
bytes = self._encode_bulk_data(body)
588
self._request.accept_bytes(bytes)
589
self._request.finished_writing()
590
self._last_verb = args[0]
592
def call_with_body_readv_array(self, args, body):
593
"""Make a remote call with a readv array.
595
The body is encoded with one line per readv offset pair. The numbers in
596
each pair are separated by a comma, and no trailing \n is emitted.
598
if 'hpss' in debug.debug_flags:
599
mutter('hpss call w/readv: %s', repr(args)[1:-1])
600
if getattr(self._request._medium, '_path', None) is not None:
601
mutter(' (to %s)', self._request._medium._path)
602
self._request_start_time = time.time()
603
self._write_args(args)
604
readv_bytes = self._serialise_offsets(body)
605
bytes = self._encode_bulk_data(readv_bytes)
606
self._request.accept_bytes(bytes)
607
self._request.finished_writing()
608
if 'hpss' in debug.debug_flags:
609
mutter(' %d bytes in readv request', len(readv_bytes))
610
self._last_verb = args[0]
612
def cancel_read_body(self):
613
"""After expecting a body, a response code may indicate one otherwise.
615
This method lets the domain client inform the protocol that no body
616
will be transmitted. This is a terminal method: after calling it the
617
protocol is not able to be used further.
619
self._request.finished_reading()
621
def _read_response_tuple(self):
622
result = self._recv_tuple()
623
if 'hpss' in debug.debug_flags:
624
if self._request_start_time is not None:
625
mutter(' result: %6.3fs %s',
626
time.time() - self._request_start_time,
628
self._request_start_time = None
630
mutter(' result: %s', repr(result)[1:-1])
633
def read_response_tuple(self, expect_body=False):
634
"""Read a response tuple from the wire.
636
This should only be called once.
638
result = self._read_response_tuple()
639
self._response_is_unknown_method(result)
640
self._raise_args_if_error(result)
642
self._request.finished_reading()
645
def _raise_args_if_error(self, result_tuple):
646
# Later protocol versions have an explicit flag in the protocol to say
647
# if an error response is "failed" or not. In version 1 we don't have
648
# that luxury. So here is a complete list of errors that can be
649
# returned in response to existing version 1 smart requests. Responses
650
# starting with these codes are always "failed" responses.
657
'UnicodeEncodeError',
658
'UnicodeDecodeError',
664
'UnlockableTransport',
670
if result_tuple[0] in v1_error_codes:
671
self._request.finished_reading()
672
raise errors.ErrorFromSmartServer(result_tuple)
674
def _response_is_unknown_method(self, result_tuple):
675
"""Raise UnexpectedSmartServerResponse if the response is an 'unknonwn
676
method' response to the request.
678
:param response: The response from a smart client call_expecting_body
680
:param verb: The verb used in that call.
681
:raises: UnexpectedSmartServerResponse
683
if (result_tuple == ('error', "Generic bzr smart protocol error: "
684
"bad request '%s'" % self._last_verb) or
685
result_tuple == ('error', "Generic bzr smart protocol error: "
686
"bad request u'%s'" % self._last_verb)):
687
# The response will have no body, so we've finished reading.
688
self._request.finished_reading()
689
raise errors.UnknownSmartMethod(self._last_verb)
691
def read_body_bytes(self, count=-1):
692
"""Read bytes from the body, decoding into a byte stream.
694
We read all bytes at once to ensure we've checked the trailer for
695
errors, and then feed the buffer back as read_body_bytes is called.
697
if self._body_buffer is not None:
698
return self._body_buffer.read(count)
699
_body_decoder = LengthPrefixedBodyDecoder()
701
# Read no more than 64k at a time so that we don't risk error 10055 (no
702
# buffer space available) on Windows.
704
while not _body_decoder.finished_reading:
705
bytes_wanted = min(_body_decoder.next_read_size(), max_read)
706
bytes = self._request.read_bytes(bytes_wanted)
708
# end of file encountered reading from server
709
raise errors.ConnectionReset(
710
"Connection lost while reading response body.")
711
_body_decoder.accept_bytes(bytes)
712
self._request.finished_reading()
713
self._body_buffer = StringIO(_body_decoder.read_pending_data())
714
# XXX: TODO check the trailer result.
715
if 'hpss' in debug.debug_flags:
716
mutter(' %d body bytes read',
717
len(self._body_buffer.getvalue()))
718
return self._body_buffer.read(count)
720
def _recv_tuple(self):
721
"""Receive a tuple from the medium request."""
722
return _decode_tuple(self._recv_line())
724
def _recv_line(self):
725
"""Read an entire line from the medium request."""
727
while not line or line[-1] != '\n':
728
# TODO: this is inefficient - but tuples are short.
729
new_char = self._request.read_bytes(1)
731
# end of file encountered reading from server
732
raise errors.ConnectionReset(
733
"please check connectivity and permissions",
734
"(and try -Dhpss if further diagnosis is required)")
738
def query_version(self):
739
"""Return protocol version number of the server."""
741
resp = self.read_response_tuple()
742
if resp == ('ok', '1'):
744
elif resp == ('ok', '2'):
747
raise errors.SmartProtocolError("bad response %r" % (resp,))
749
def _write_args(self, args):
750
self._write_protocol_version()
751
bytes = _encode_tuple(args)
752
self._request.accept_bytes(bytes)
754
def _write_protocol_version(self):
755
"""Write any prefixes this protocol requires.
757
Version one doesn't send protocol versions.
761
class SmartClientRequestProtocolTwo(SmartClientRequestProtocolOne):
762
"""Version two of the client side of the smart protocol.
764
This prefixes the request with the value of REQUEST_VERSION_TWO.
767
response_marker = RESPONSE_VERSION_TWO
768
request_marker = REQUEST_VERSION_TWO
770
def read_response_tuple(self, expect_body=False):
771
"""Read a response tuple from the wire.
773
This should only be called once.
775
version = self._request.read_line()
776
if version != self.response_marker:
777
self._request.finished_reading()
778
raise errors.UnexpectedProtocolVersionMarker(version)
779
response_status = self._recv_line()
780
result = SmartClientRequestProtocolOne._read_response_tuple(self)
781
self._response_is_unknown_method(result)
782
if response_status == 'success\n':
783
self.response_status = True
785
self._request.finished_reading()
787
elif response_status == 'failed\n':
788
self.response_status = False
789
self._request.finished_reading()
790
raise errors.ErrorFromSmartServer(result)
792
raise errors.SmartProtocolError(
793
'bad protocol status %r' % response_status)
795
def _write_protocol_version(self):
796
"""Write any prefixes this protocol requires.
798
Version two sends the value of REQUEST_VERSION_TWO.
800
self._request.accept_bytes(self.request_marker)
802
def read_streamed_body(self):
803
"""Read bytes from the body, decoding into a byte stream.
805
# Read no more than 64k at a time so that we don't risk error 10055 (no
806
# buffer space available) on Windows.
808
_body_decoder = ChunkedBodyDecoder()
809
while not _body_decoder.finished_reading:
810
bytes_wanted = min(_body_decoder.next_read_size(), max_read)
811
bytes = self._request.read_bytes(bytes_wanted)
813
# end of file encountered reading from server
814
raise errors.ConnectionReset(
815
"Connection lost while reading streamed body.")
816
_body_decoder.accept_bytes(bytes)
817
for body_bytes in iter(_body_decoder.read_next_chunk, None):
818
if 'hpss' in debug.debug_flags and type(body_bytes) is str:
819
mutter(' %d byte chunk read',
822
self._request.finished_reading()
825
def build_server_protocol_three(backing_transport, write_func,
827
request_handler = request.SmartServerRequestHandler(
828
backing_transport, commands=request.request_handlers,
829
root_client_path=root_client_path)
830
responder = ProtocolThreeResponder(write_func)
831
message_handler = message.ConventionalRequestHandler(request_handler, responder)
832
return ProtocolThreeDecoder(message_handler)
835
class ProtocolThreeDecoder(_StatefulDecoder):
837
response_marker = RESPONSE_VERSION_THREE
838
request_marker = REQUEST_VERSION_THREE
840
def __init__(self, message_handler, expect_version_marker=False):
841
_StatefulDecoder.__init__(self)
842
self._has_dispatched = False
844
if expect_version_marker:
845
self.state_accept = self._state_accept_expecting_protocol_version
846
# We're expecting at least the protocol version marker + some
848
self._number_needed_bytes = len(MESSAGE_VERSION_THREE) + 4
850
self.state_accept = self._state_accept_expecting_headers
851
self._number_needed_bytes = 4
852
self.decoding_failed = False
853
self.request_handler = self.message_handler = message_handler
855
def accept_bytes(self, bytes):
856
self._number_needed_bytes = None
858
_StatefulDecoder.accept_bytes(self, bytes)
859
except KeyboardInterrupt:
861
except errors.SmartMessageHandlerError, exception:
862
# We do *not* set self.decoding_failed here. The message handler
863
# has raised an error, but the decoder is still able to parse bytes
864
# and determine when this message ends.
865
log_exception_quietly()
866
self.message_handler.protocol_error(exception.exc_value)
867
# The state machine is ready to continue decoding, but the
868
# exception has interrupted the loop that runs the state machine.
869
# So we call accept_bytes again to restart it.
870
self.accept_bytes('')
871
except Exception, exception:
872
# The decoder itself has raised an exception. We cannot continue
874
self.decoding_failed = True
875
if isinstance(exception, errors.UnexpectedProtocolVersionMarker):
876
# This happens during normal operation when the client tries a
877
# protocol version the server doesn't understand, so no need to
878
# log a traceback every time.
879
# Note that this can only happen when
880
# expect_version_marker=True, which is only the case on the
884
log_exception_quietly()
885
self.message_handler.protocol_error(exception)
887
def _extract_length_prefixed_bytes(self):
888
if len(self._in_buffer) < 4:
889
# A length prefix by itself is 4 bytes, and we don't even have that
891
raise _NeedMoreBytes(4)
892
(length,) = struct.unpack('!L', self._in_buffer[:4])
893
end_of_bytes = 4 + length
894
if len(self._in_buffer) < end_of_bytes:
895
# We haven't yet read as many bytes as the length-prefix says there
897
raise _NeedMoreBytes(end_of_bytes)
898
# Extract the bytes from the buffer.
899
bytes = self._in_buffer[4:end_of_bytes]
900
self._in_buffer = self._in_buffer[end_of_bytes:]
903
def _extract_prefixed_bencoded_data(self):
904
prefixed_bytes = self._extract_length_prefixed_bytes()
906
decoded = bdecode(prefixed_bytes)
908
raise errors.SmartProtocolError(
909
'Bytes %r not bencoded' % (prefixed_bytes,))
912
def _extract_single_byte(self):
913
if self._in_buffer == '':
914
# The buffer is empty
915
raise _NeedMoreBytes(1)
916
one_byte = self._in_buffer[0]
917
self._in_buffer = self._in_buffer[1:]
920
def _state_accept_expecting_protocol_version(self):
921
needed_bytes = len(MESSAGE_VERSION_THREE) - len(self._in_buffer)
923
# We don't have enough bytes to check if the protocol version
924
# marker is right. But we can check if it is already wrong by
925
# checking that the start of MESSAGE_VERSION_THREE matches what
927
# [In fact, if the remote end isn't bzr we might never receive
928
# len(MESSAGE_VERSION_THREE) bytes. So if the bytes we have so far
929
# are wrong then we should just raise immediately rather than
931
if not MESSAGE_VERSION_THREE.startswith(self._in_buffer):
932
# We have enough bytes to know the protocol version is wrong
933
raise errors.UnexpectedProtocolVersionMarker(self._in_buffer)
934
raise _NeedMoreBytes(len(MESSAGE_VERSION_THREE))
935
if not self._in_buffer.startswith(MESSAGE_VERSION_THREE):
936
raise errors.UnexpectedProtocolVersionMarker(self._in_buffer)
937
self._in_buffer = self._in_buffer[len(MESSAGE_VERSION_THREE):]
938
self.state_accept = self._state_accept_expecting_headers
940
def _state_accept_expecting_headers(self):
941
decoded = self._extract_prefixed_bencoded_data()
942
if type(decoded) is not dict:
943
raise errors.SmartProtocolError(
944
'Header object %r is not a dict' % (decoded,))
945
self.state_accept = self._state_accept_expecting_message_part
947
self.message_handler.headers_received(decoded)
949
raise errors.SmartMessageHandlerError(sys.exc_info())
951
def _state_accept_expecting_message_part(self):
952
message_part_kind = self._extract_single_byte()
953
if message_part_kind == 'o':
954
self.state_accept = self._state_accept_expecting_one_byte
955
elif message_part_kind == 's':
956
self.state_accept = self._state_accept_expecting_structure
957
elif message_part_kind == 'b':
958
self.state_accept = self._state_accept_expecting_bytes
959
elif message_part_kind == 'e':
962
raise errors.SmartProtocolError(
963
'Bad message kind byte: %r' % (message_part_kind,))
965
def _state_accept_expecting_one_byte(self):
966
byte = self._extract_single_byte()
967
self.state_accept = self._state_accept_expecting_message_part
969
self.message_handler.byte_part_received(byte)
971
raise errors.SmartMessageHandlerError(sys.exc_info())
973
def _state_accept_expecting_bytes(self):
974
# XXX: this should not buffer whole message part, but instead deliver
975
# the bytes as they arrive.
976
prefixed_bytes = self._extract_length_prefixed_bytes()
977
self.state_accept = self._state_accept_expecting_message_part
979
self.message_handler.bytes_part_received(prefixed_bytes)
981
raise errors.SmartMessageHandlerError(sys.exc_info())
983
def _state_accept_expecting_structure(self):
984
structure = self._extract_prefixed_bencoded_data()
985
self.state_accept = self._state_accept_expecting_message_part
987
self.message_handler.structure_part_received(structure)
989
raise errors.SmartMessageHandlerError(sys.exc_info())
992
self.unused_data = self._in_buffer
994
self.state_accept = self._state_accept_reading_unused
996
self.message_handler.end_received()
998
raise errors.SmartMessageHandlerError(sys.exc_info())
1000
def _state_accept_reading_unused(self):
1001
self.unused_data += self._in_buffer
1002
self._in_buffer = ''
1004
def next_read_size(self):
1005
if self.state_accept == self._state_accept_reading_unused:
1007
elif self.decoding_failed:
1008
# An exception occured while processing this message, probably from
1009
# self.message_handler. We're not sure that this state machine is
1010
# in a consistent state, so just signal that we're done (i.e. give
1014
if self._number_needed_bytes is not None:
1015
return self._number_needed_bytes - len(self._in_buffer)
1017
raise AssertionError("don't know how many bytes are expected!")
1020
class _ProtocolThreeEncoder(object):
1022
response_marker = request_marker = MESSAGE_VERSION_THREE
1024
def __init__(self, write_func):
1026
self._real_write_func = write_func
1028
def _write_func(self, bytes):
1033
self._real_write_func(self._buf)
1036
def _serialise_offsets(self, offsets):
1037
"""Serialise a readv offset list."""
1039
for start, length in offsets:
1040
txt.append('%d,%d' % (start, length))
1041
return '\n'.join(txt)
1043
def _write_protocol_version(self):
1044
self._write_func(MESSAGE_VERSION_THREE)
1046
def _write_prefixed_bencode(self, structure):
1047
bytes = bencode(structure)
1048
self._write_func(struct.pack('!L', len(bytes)))
1049
self._write_func(bytes)
1051
def _write_headers(self, headers):
1052
self._write_prefixed_bencode(headers)
1054
def _write_structure(self, args):
1055
self._write_func('s')
1058
if type(arg) is unicode:
1059
utf8_args.append(arg.encode('utf8'))
1061
utf8_args.append(arg)
1062
self._write_prefixed_bencode(utf8_args)
1064
def _write_end(self):
1065
self._write_func('e')
1068
def _write_prefixed_body(self, bytes):
1069
self._write_func('b')
1070
self._write_func(struct.pack('!L', len(bytes)))
1071
self._write_func(bytes)
1073
def _write_error_status(self):
1074
self._write_func('oE')
1076
def _write_success_status(self):
1077
self._write_func('oS')
1080
class ProtocolThreeResponder(_ProtocolThreeEncoder):
1082
def __init__(self, write_func):
1083
_ProtocolThreeEncoder.__init__(self, write_func)
1084
self.response_sent = False
1085
self._headers = {'Software version': bzrlib.__version__}
1087
def send_error(self, exception):
1088
if self.response_sent:
1089
raise AssertionError(
1090
"send_error(%s) called, but response already sent."
1092
if isinstance(exception, errors.UnknownSmartMethod):
1093
failure = request.FailedSmartServerResponse(
1094
('UnknownMethod', exception.verb))
1095
self.send_response(failure)
1097
self.response_sent = True
1098
self._write_protocol_version()
1099
self._write_headers(self._headers)
1100
self._write_error_status()
1101
self._write_structure(('error', str(exception)))
1104
def send_response(self, response):
1105
if self.response_sent:
1106
raise AssertionError(
1107
"send_response(%r) called, but response already sent."
1109
self.response_sent = True
1110
self._write_protocol_version()
1111
self._write_headers(self._headers)
1112
if response.is_successful():
1113
self._write_success_status()
1115
self._write_error_status()
1116
self._write_structure(response.args)
1117
if response.body is not None:
1118
self._write_prefixed_body(response.body)
1119
elif response.body_stream is not None:
1120
for chunk in response.body_stream:
1121
self._write_prefixed_body(chunk)
1126
class ProtocolThreeRequester(_ProtocolThreeEncoder, Requester):
1128
def __init__(self, medium_request):
1129
_ProtocolThreeEncoder.__init__(self, medium_request.accept_bytes)
1130
self._medium_request = medium_request
1133
def set_headers(self, headers):
1134
self._headers = headers.copy()
1136
def call(self, *args):
1137
if 'hpss' in debug.debug_flags:
1138
mutter('hpss call: %s', repr(args)[1:-1])
1139
base = getattr(self._medium_request._medium, 'base', None)
1140
if base is not None:
1141
mutter(' (to %s)', base)
1142
self._request_start_time = time.time()
1143
self._write_protocol_version()
1144
self._write_headers(self._headers)
1145
self._write_structure(args)
1147
self._medium_request.finished_writing()
1149
def call_with_body_bytes(self, args, body):
1150
"""Make a remote call of args with body bytes 'body'.
1152
After calling this, call read_response_tuple to find the result out.
1154
if 'hpss' in debug.debug_flags:
1155
mutter('hpss call w/body: %s (%r...)', repr(args)[1:-1], body[:20])
1156
path = getattr(self._medium_request._medium, '_path', None)
1157
if path is not None:
1158
mutter(' (to %s)', path)
1159
mutter(' %d bytes', len(body))
1160
self._request_start_time = time.time()
1161
self._write_protocol_version()
1162
self._write_headers(self._headers)
1163
self._write_structure(args)
1164
self._write_prefixed_body(body)
1166
self._medium_request.finished_writing()
1168
def call_with_body_readv_array(self, args, body):
1169
"""Make a remote call with a readv array.
1171
The body is encoded with one line per readv offset pair. The numbers in
1172
each pair are separated by a comma, and no trailing \n is emitted.
1174
if 'hpss' in debug.debug_flags:
1175
mutter('hpss call w/readv: %s', repr(args)[1:-1])
1176
path = getattr(self._medium_request._medium, '_path', None)
1177
if path is not None:
1178
mutter(' (to %s)', path)
1179
self._request_start_time = time.time()
1180
self._write_protocol_version()
1181
self._write_headers(self._headers)
1182
self._write_structure(args)
1183
readv_bytes = self._serialise_offsets(body)
1184
if 'hpss' in debug.debug_flags:
1185
mutter(' %d bytes in readv request', len(readv_bytes))
1186
self._write_prefixed_body(readv_bytes)
1188
self._medium_request.finished_writing()
added
removed
bzrlib/smart/protocol.py
