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
21
from cStringIO import StringIO
24
from bzrlib import debug
25
from bzrlib import errors
26
from bzrlib.smart import request
27
from bzrlib.trace import log_exception_quietly, mutter
30
# Protocol version strings. These are sent as prefixes of bzr requests and
31
# responses to identify the protocol version being used. (There are no version
32
# one strings because that version doesn't send any).
33
REQUEST_VERSION_TWO = 'bzr request 2\n'
34
RESPONSE_VERSION_TWO = 'bzr response 2\n'
37
def _recv_tuple(from_file):
38
req_line = from_file.readline()
39
return _decode_tuple(req_line)
42
def _decode_tuple(req_line):
43
if req_line == None or req_line == '':
45
if req_line[-1] != '\n':
46
raise errors.SmartProtocolError("request %r not terminated" % req_line)
47
return tuple(req_line[:-1].split('\x01'))
50
def _encode_tuple(args):
51
"""Encode the tuple args to a bytestream."""
52
return '\x01'.join(args) + '\n'
55
class SmartProtocolBase(object):
56
"""Methods common to client and server"""
58
# TODO: this only actually accomodates a single block; possibly should
59
# support multiple chunks?
60
def _encode_bulk_data(self, body):
61
"""Encode body as a bulk data chunk."""
62
return ''.join(('%d\n' % len(body), body, 'done\n'))
64
def _serialise_offsets(self, offsets):
65
"""Serialise a readv offset list."""
67
for start, length in offsets:
68
txt.append('%d,%d' % (start, length))
72
class SmartServerRequestProtocolOne(SmartProtocolBase):
73
"""Server-side encoding and decoding logic for smart version 1."""
75
def __init__(self, backing_transport, write_func):
76
self._backing_transport = backing_transport
77
self.excess_buffer = ''
78
self._finished = False
80
self.has_dispatched = False
82
self._body_decoder = None
83
self._write_func = write_func
85
def accept_bytes(self, bytes):
86
"""Take bytes, and advance the internal state machine appropriately.
88
:param bytes: must be a byte string
90
assert isinstance(bytes, str)
91
self.in_buffer += bytes
92
if not self.has_dispatched:
93
if '\n' not in self.in_buffer:
96
self.has_dispatched = True
98
first_line, self.in_buffer = self.in_buffer.split('\n', 1)
100
req_args = _decode_tuple(first_line)
101
self.request = request.SmartServerRequestHandler(
102
self._backing_transport, commands=request.request_handlers)
103
self.request.dispatch_command(req_args[0], req_args[1:])
104
if self.request.finished_reading:
106
self.excess_buffer = self.in_buffer
108
self._send_response(self.request.response)
109
except KeyboardInterrupt:
111
except Exception, exception:
112
# everything else: pass to client, flush, and quit
113
log_exception_quietly()
114
self._send_response(request.FailedSmartServerResponse(
115
('error', str(exception))))
118
if self.has_dispatched:
120
# nothing to do.XXX: this routine should be a single state
122
self.excess_buffer += self.in_buffer
125
if self._body_decoder is None:
126
self._body_decoder = LengthPrefixedBodyDecoder()
127
self._body_decoder.accept_bytes(self.in_buffer)
128
self.in_buffer = self._body_decoder.unused_data
129
body_data = self._body_decoder.read_pending_data()
130
self.request.accept_body(body_data)
131
if self._body_decoder.finished_reading:
132
self.request.end_of_body()
133
assert self.request.finished_reading, \
134
"no more body, request not finished"
135
if self.request.response is not None:
136
self._send_response(self.request.response)
137
self.excess_buffer = self.in_buffer
140
assert not self.request.finished_reading, \
141
"no response and we have finished reading."
143
def _send_response(self, response):
144
"""Send a smart server response down the output stream."""
145
assert not self._finished, 'response already sent'
148
self._finished = True
149
self._write_protocol_version()
150
self._write_success_or_failure_prefix(response)
151
self._write_func(_encode_tuple(args))
153
assert isinstance(body, str), 'body must be a str'
154
bytes = self._encode_bulk_data(body)
155
self._write_func(bytes)
157
def _write_protocol_version(self):
158
"""Write any prefixes this protocol requires.
160
Version one doesn't send protocol versions.
163
def _write_success_or_failure_prefix(self, response):
164
"""Write the protocol specific success/failure prefix.
166
For SmartServerRequestProtocolOne this is omitted but we
167
call is_successful to ensure that the response is valid.
169
response.is_successful()
171
def next_read_size(self):
174
if self._body_decoder is None:
177
return self._body_decoder.next_read_size()
180
class SmartServerRequestProtocolTwo(SmartServerRequestProtocolOne):
181
r"""Version two of the server side of the smart protocol.
183
This prefixes responses with the value of RESPONSE_VERSION_TWO.
186
def _write_success_or_failure_prefix(self, response):
187
"""Write the protocol specific success/failure prefix."""
188
if response.is_successful():
189
self._write_func('success\n')
191
self._write_func('failed\n')
193
def _write_protocol_version(self):
194
r"""Write any prefixes this protocol requires.
196
Version two sends the value of RESPONSE_VERSION_TWO.
198
self._write_func(RESPONSE_VERSION_TWO)
201
class LengthPrefixedBodyDecoder(object):
202
"""Decodes the length-prefixed bulk data."""
205
self.bytes_left = None
206
self.finished_reading = False
207
self.unused_data = ''
208
self.state_accept = self._state_accept_expecting_length
209
self.state_read = self._state_read_no_data
211
self._trailer_buffer = ''
213
def accept_bytes(self, bytes):
214
"""Decode as much of bytes as possible.
216
If 'bytes' contains too much data it will be appended to
219
finished_reading will be set when no more data is required. Further
220
data will be appended to self.unused_data.
222
# accept_bytes is allowed to change the state
223
current_state = self.state_accept
224
self.state_accept(bytes)
225
while current_state != self.state_accept:
226
current_state = self.state_accept
227
self.state_accept('')
229
def next_read_size(self):
230
if self.bytes_left is not None:
231
# Ideally we want to read all the remainder of the body and the
233
return self.bytes_left + 5
234
elif self.state_accept == self._state_accept_reading_trailer:
235
# Just the trailer left
236
return 5 - len(self._trailer_buffer)
237
elif self.state_accept == self._state_accept_expecting_length:
238
# There's still at least 6 bytes left ('\n' to end the length, plus
242
# Reading excess data. Either way, 1 byte at a time is fine.
245
def read_pending_data(self):
246
"""Return any pending data that has been decoded."""
247
return self.state_read()
249
def _state_accept_expecting_length(self, bytes):
250
self._in_buffer += bytes
251
pos = self._in_buffer.find('\n')
254
self.bytes_left = int(self._in_buffer[:pos])
255
self._in_buffer = self._in_buffer[pos+1:]
256
self.bytes_left -= len(self._in_buffer)
257
self.state_accept = self._state_accept_reading_body
258
self.state_read = self._state_read_in_buffer
260
def _state_accept_reading_body(self, bytes):
261
self._in_buffer += bytes
262
self.bytes_left -= len(bytes)
263
if self.bytes_left <= 0:
265
if self.bytes_left != 0:
266
self._trailer_buffer = self._in_buffer[self.bytes_left:]
267
self._in_buffer = self._in_buffer[:self.bytes_left]
268
self.bytes_left = None
269
self.state_accept = self._state_accept_reading_trailer
271
def _state_accept_reading_trailer(self, bytes):
272
self._trailer_buffer += bytes
273
# TODO: what if the trailer does not match "done\n"? Should this raise
274
# a ProtocolViolation exception?
275
if self._trailer_buffer.startswith('done\n'):
276
self.unused_data = self._trailer_buffer[len('done\n'):]
277
self.state_accept = self._state_accept_reading_unused
278
self.finished_reading = True
280
def _state_accept_reading_unused(self, bytes):
281
self.unused_data += bytes
283
def _state_read_no_data(self):
286
def _state_read_in_buffer(self):
287
result = self._in_buffer
292
class SmartClientRequestProtocolOne(SmartProtocolBase):
293
"""The client-side protocol for smart version 1."""
295
def __init__(self, request):
296
"""Construct a SmartClientRequestProtocolOne.
298
:param request: A SmartClientMediumRequest to serialise onto and
301
self._request = request
302
self._body_buffer = None
303
self._request_start_time = None
305
def call(self, *args):
306
if 'hpss' in debug.debug_flags:
307
mutter('hpss call: %s', repr(args)[1:-1])
308
self._request_start_time = time.time()
309
self._write_args(args)
310
self._request.finished_writing()
312
def call_with_body_bytes(self, args, body):
313
"""Make a remote call of args with body bytes 'body'.
315
After calling this, call read_response_tuple to find the result out.
317
if 'hpss' in debug.debug_flags:
318
mutter('hpss call w/body: %s (%r...)', repr(args)[1:-1], body[:20])
319
mutter(' %d bytes', len(body))
320
self._request_start_time = time.time()
321
self._write_args(args)
322
bytes = self._encode_bulk_data(body)
323
self._request.accept_bytes(bytes)
324
self._request.finished_writing()
326
def call_with_body_readv_array(self, args, body):
327
"""Make a remote call with a readv array.
329
The body is encoded with one line per readv offset pair. The numbers in
330
each pair are separated by a comma, and no trailing \n is emitted.
332
if 'hpss' in debug.debug_flags:
333
mutter('hpss call w/readv: %s', repr(args)[1:-1])
334
self._request_start_time = time.time()
335
self._write_args(args)
336
readv_bytes = self._serialise_offsets(body)
337
bytes = self._encode_bulk_data(readv_bytes)
338
self._request.accept_bytes(bytes)
339
self._request.finished_writing()
340
if 'hpss' in debug.debug_flags:
341
mutter(' %d bytes in readv request', len(readv_bytes))
343
def cancel_read_body(self):
344
"""After expecting a body, a response code may indicate one otherwise.
346
This method lets the domain client inform the protocol that no body
347
will be transmitted. This is a terminal method: after calling it the
348
protocol is not able to be used further.
350
self._request.finished_reading()
352
def read_response_tuple(self, expect_body=False):
353
"""Read a response tuple from the wire.
355
This should only be called once.
357
result = self._recv_tuple()
358
if 'hpss' in debug.debug_flags:
359
if self._request_start_time is not None:
360
mutter(' result: %6.3fs %s',
361
time.time() - self._request_start_time,
363
self._request_start_time = None
365
mutter(' result: %s', repr(result)[1:-1])
367
self._request.finished_reading()
370
def read_body_bytes(self, count=-1):
371
"""Read bytes from the body, decoding into a byte stream.
373
We read all bytes at once to ensure we've checked the trailer for
374
errors, and then feed the buffer back as read_body_bytes is called.
376
if self._body_buffer is not None:
377
return self._body_buffer.read(count)
378
_body_decoder = LengthPrefixedBodyDecoder()
380
while not _body_decoder.finished_reading:
381
bytes_wanted = _body_decoder.next_read_size()
382
bytes = self._request.read_bytes(bytes_wanted)
383
_body_decoder.accept_bytes(bytes)
384
self._request.finished_reading()
385
self._body_buffer = StringIO(_body_decoder.read_pending_data())
386
# XXX: TODO check the trailer result.
387
if 'hpss' in debug.debug_flags:
388
mutter(' %d body bytes read',
389
len(self._body_buffer.getvalue()))
390
return self._body_buffer.read(count)
392
def _recv_tuple(self):
393
"""Receive a tuple from the medium request."""
394
return _decode_tuple(self._recv_line())
396
def _recv_line(self):
397
"""Read an entire line from the medium request."""
399
while not line or line[-1] != '\n':
400
# TODO: this is inefficient - but tuples are short.
401
new_char = self._request.read_bytes(1)
403
assert new_char != '', "end of file reading from server."
406
def query_version(self):
407
"""Return protocol version number of the server."""
409
resp = self.read_response_tuple()
410
if resp == ('ok', '1'):
412
elif resp == ('ok', '2'):
415
raise errors.SmartProtocolError("bad response %r" % (resp,))
417
def _write_args(self, args):
418
self._write_protocol_version()
419
bytes = _encode_tuple(args)
420
self._request.accept_bytes(bytes)
422
def _write_protocol_version(self):
423
"""Write any prefixes this protocol requires.
425
Version one doesn't send protocol versions.
429
class SmartClientRequestProtocolTwo(SmartClientRequestProtocolOne):
430
"""Version two of the client side of the smart protocol.
432
This prefixes the request with the value of REQUEST_VERSION_TWO.
435
def read_response_tuple(self, expect_body=False):
436
"""Read a response tuple from the wire.
438
This should only be called once.
440
version = self._request.read_line()
441
if version != RESPONSE_VERSION_TWO:
442
raise errors.SmartProtocolError('bad protocol marker %r' % version)
443
response_status = self._recv_line()
444
if response_status not in ('success\n', 'failed\n'):
445
raise errors.SmartProtocolError(
446
'bad protocol status %r' % response_status)
447
self.response_status = response_status == 'success\n'
448
return SmartClientRequestProtocolOne.read_response_tuple(self, expect_body)
450
def _write_protocol_version(self):
451
r"""Write any prefixes this protocol requires.
453
Version two sends the value of REQUEST_VERSION_TWO.
455
self._request.accept_bytes(REQUEST_VERSION_TWO)
added
removed
bzrlib/smart/protocol.py
