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 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
self._send_response(request.FailedSmartServerResponse(
114
('error', str(exception))))
117
if self.has_dispatched:
119
# nothing to do.XXX: this routine should be a single state
121
self.excess_buffer += self.in_buffer
124
if self._body_decoder is None:
125
self._body_decoder = LengthPrefixedBodyDecoder()
126
self._body_decoder.accept_bytes(self.in_buffer)
127
self.in_buffer = self._body_decoder.unused_data
128
body_data = self._body_decoder.read_pending_data()
129
self.request.accept_body(body_data)
130
if self._body_decoder.finished_reading:
131
self.request.end_of_body()
132
assert self.request.finished_reading, \
133
"no more body, request not finished"
134
if self.request.response is not None:
135
self._send_response(self.request.response)
136
self.excess_buffer = self.in_buffer
139
assert not self.request.finished_reading, \
140
"no response and we have finished reading."
142
def _send_response(self, response):
143
"""Send a smart server response down the output stream."""
144
assert not self._finished, 'response already sent'
147
self._finished = True
148
self._write_protocol_version()
149
self._write_success_or_failure_prefix(response)
150
self._write_func(_encode_tuple(args))
152
assert isinstance(body, str), 'body must be a str'
153
bytes = self._encode_bulk_data(body)
154
self._write_func(bytes)
156
def _write_protocol_version(self):
157
"""Write any prefixes this protocol requires.
159
Version one doesn't send protocol versions.
162
def _write_success_or_failure_prefix(self, response):
163
"""Write the protocol specific success/failure prefix.
165
For SmartServerRequestProtocolOne this is omitted but we
166
call is_successful to ensure that the response is valid.
168
response.is_successful()
170
def next_read_size(self):
173
if self._body_decoder is None:
176
return self._body_decoder.next_read_size()
179
class SmartServerRequestProtocolTwo(SmartServerRequestProtocolOne):
180
r"""Version two of the server side of the smart protocol.
182
This prefixes responses with the value of RESPONSE_VERSION_TWO.
185
def _write_success_or_failure_prefix(self, response):
186
"""Write the protocol specific success/failure prefix."""
187
if response.is_successful():
188
self._write_func('success\n')
190
self._write_func('failed\n')
192
def _write_protocol_version(self):
193
r"""Write any prefixes this protocol requires.
195
Version two sends the value of RESPONSE_VERSION_TWO.
197
self._write_func(RESPONSE_VERSION_TWO)
200
class LengthPrefixedBodyDecoder(object):
201
"""Decodes the length-prefixed bulk data."""
204
self.bytes_left = None
205
self.finished_reading = False
206
self.unused_data = ''
207
self.state_accept = self._state_accept_expecting_length
208
self.state_read = self._state_read_no_data
210
self._trailer_buffer = ''
212
def accept_bytes(self, bytes):
213
"""Decode as much of bytes as possible.
215
If 'bytes' contains too much data it will be appended to
218
finished_reading will be set when no more data is required. Further
219
data will be appended to self.unused_data.
221
# accept_bytes is allowed to change the state
222
current_state = self.state_accept
223
self.state_accept(bytes)
224
while current_state != self.state_accept:
225
current_state = self.state_accept
226
self.state_accept('')
228
def next_read_size(self):
229
if self.bytes_left is not None:
230
# Ideally we want to read all the remainder of the body and the
232
return self.bytes_left + 5
233
elif self.state_accept == self._state_accept_reading_trailer:
234
# Just the trailer left
235
return 5 - len(self._trailer_buffer)
236
elif self.state_accept == self._state_accept_expecting_length:
237
# There's still at least 6 bytes left ('\n' to end the length, plus
241
# Reading excess data. Either way, 1 byte at a time is fine.
244
def read_pending_data(self):
245
"""Return any pending data that has been decoded."""
246
return self.state_read()
248
def _state_accept_expecting_length(self, bytes):
249
self._in_buffer += bytes
250
pos = self._in_buffer.find('\n')
253
self.bytes_left = int(self._in_buffer[:pos])
254
self._in_buffer = self._in_buffer[pos+1:]
255
self.bytes_left -= len(self._in_buffer)
256
self.state_accept = self._state_accept_reading_body
257
self.state_read = self._state_read_in_buffer
259
def _state_accept_reading_body(self, bytes):
260
self._in_buffer += bytes
261
self.bytes_left -= len(bytes)
262
if self.bytes_left <= 0:
264
if self.bytes_left != 0:
265
self._trailer_buffer = self._in_buffer[self.bytes_left:]
266
self._in_buffer = self._in_buffer[:self.bytes_left]
267
self.bytes_left = None
268
self.state_accept = self._state_accept_reading_trailer
270
def _state_accept_reading_trailer(self, bytes):
271
self._trailer_buffer += bytes
272
# TODO: what if the trailer does not match "done\n"? Should this raise
273
# a ProtocolViolation exception?
274
if self._trailer_buffer.startswith('done\n'):
275
self.unused_data = self._trailer_buffer[len('done\n'):]
276
self.state_accept = self._state_accept_reading_unused
277
self.finished_reading = True
279
def _state_accept_reading_unused(self, bytes):
280
self.unused_data += bytes
282
def _state_read_no_data(self):
285
def _state_read_in_buffer(self):
286
result = self._in_buffer
291
class SmartClientRequestProtocolOne(SmartProtocolBase):
292
"""The client-side protocol for smart version 1."""
294
def __init__(self, request):
295
"""Construct a SmartClientRequestProtocolOne.
297
:param request: A SmartClientMediumRequest to serialise onto and
300
self._request = request
301
self._body_buffer = None
302
self._request_start_time = None
304
def call(self, *args):
305
if 'hpss' in debug.debug_flags:
306
mutter('hpss call: %s', repr(args)[1:-1])
307
self._request_start_time = time.time()
308
self._write_args(args)
309
self._request.finished_writing()
311
def call_with_body_bytes(self, args, body):
312
"""Make a remote call of args with body bytes 'body'.
314
After calling this, call read_response_tuple to find the result out.
316
if 'hpss' in debug.debug_flags:
317
mutter('hpss call w/body: %s (%r...)', repr(args)[1:-1], body[:20])
318
mutter(' %d bytes', len(body))
319
self._request_start_time = time.time()
320
self._write_args(args)
321
bytes = self._encode_bulk_data(body)
322
self._request.accept_bytes(bytes)
323
self._request.finished_writing()
325
def call_with_body_readv_array(self, args, body):
326
"""Make a remote call with a readv array.
328
The body is encoded with one line per readv offset pair. The numbers in
329
each pair are separated by a comma, and no trailing \n is emitted.
331
if 'hpss' in debug.debug_flags:
332
mutter('hpss call w/readv: %s', repr(args)[1:-1])
333
self._request_start_time = time.time()
334
self._write_args(args)
335
readv_bytes = self._serialise_offsets(body)
336
bytes = self._encode_bulk_data(readv_bytes)
337
self._request.accept_bytes(bytes)
338
self._request.finished_writing()
339
if 'hpss' in debug.debug_flags:
340
mutter(' %d bytes in readv request', len(readv_bytes))
342
def cancel_read_body(self):
343
"""After expecting a body, a response code may indicate one otherwise.
345
This method lets the domain client inform the protocol that no body
346
will be transmitted. This is a terminal method: after calling it the
347
protocol is not able to be used further.
349
self._request.finished_reading()
351
def read_response_tuple(self, expect_body=False):
352
"""Read a response tuple from the wire.
354
This should only be called once.
356
result = self._recv_tuple()
357
if 'hpss' in debug.debug_flags:
358
if self._request_start_time is not None:
359
mutter(' result: %6.3fs %s',
360
time.time() - self._request_start_time,
362
self._request_start_time = None
364
mutter(' result: %s', repr(result)[1:-1])
366
self._request.finished_reading()
369
def read_body_bytes(self, count=-1):
370
"""Read bytes from the body, decoding into a byte stream.
372
We read all bytes at once to ensure we've checked the trailer for
373
errors, and then feed the buffer back as read_body_bytes is called.
375
if self._body_buffer is not None:
376
return self._body_buffer.read(count)
377
_body_decoder = LengthPrefixedBodyDecoder()
379
while not _body_decoder.finished_reading:
380
bytes_wanted = _body_decoder.next_read_size()
381
bytes = self._request.read_bytes(bytes_wanted)
382
_body_decoder.accept_bytes(bytes)
383
self._request.finished_reading()
384
self._body_buffer = StringIO(_body_decoder.read_pending_data())
385
# XXX: TODO check the trailer result.
386
if 'hpss' in debug.debug_flags:
387
mutter(' %d body bytes read',
388
len(self._body_buffer.getvalue()))
389
return self._body_buffer.read(count)
391
def _recv_tuple(self):
392
"""Receive a tuple from the medium request."""
393
return _decode_tuple(self._recv_line())
395
def _recv_line(self):
396
"""Read an entire line from the medium request."""
398
while not line or line[-1] != '\n':
399
# TODO: this is inefficient - but tuples are short.
400
new_char = self._request.read_bytes(1)
402
assert new_char != '', "end of file reading from server."
405
def query_version(self):
406
"""Return protocol version number of the server."""
408
resp = self.read_response_tuple()
409
if resp == ('ok', '1'):
411
elif resp == ('ok', '2'):
414
raise errors.SmartProtocolError("bad response %r" % (resp,))
416
def _write_args(self, args):
417
self._write_protocol_version()
418
bytes = _encode_tuple(args)
419
self._request.accept_bytes(bytes)
421
def _write_protocol_version(self):
422
"""Write any prefixes this protocol requires.
424
Version one doesn't send protocol versions.
428
class SmartClientRequestProtocolTwo(SmartClientRequestProtocolOne):
429
"""Version two of the client side of the smart protocol.
431
This prefixes the request with the value of REQUEST_VERSION_TWO.
434
def read_response_tuple(self, expect_body=False):
435
"""Read a response tuple from the wire.
437
This should only be called once.
439
version = self._request.read_line()
440
if version != RESPONSE_VERSION_TWO:
441
raise errors.SmartProtocolError('bad protocol marker %r' % version)
442
response_status = self._recv_line()
443
if response_status not in ('success\n', 'failed\n'):
444
raise errors.SmartProtocolError(
445
'bad protocol status %r' % response_status)
446
self.response_status = response_status == 'success\n'
447
return SmartClientRequestProtocolOne.read_response_tuple(self, expect_body)
449
def _write_protocol_version(self):
450
r"""Write any prefixes this protocol requires.
452
Version two sends the value of REQUEST_VERSION_TWO.
454
self._request.accept_bytes(REQUEST_VERSION_TWO)
added
removed
bzrlib/smart/protocol.py
