/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/smart/protocol.py

  • Committer: Robert Collins
  • Date: 2007-04-24 12:20:09 UTC
  • mto: (2432.3.5 hpss-vfs-fallback)
  • mto: This revision was merged to the branch mainline in revision 2463.
  • Revision ID: robertc@lifelesswks.robertcollins.net-20070424122009-8bb4dede6a298d93
Make using SuccessfulSmartServerResponse and FailedSmartServerResponse mandatory rather than optional in smart server logic.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/protocol.py
 
1
# Copyright (C) 2006, 2007 Canonical Ltd
 
2
#
 
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.
 
7
#
 
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.
 
12
#
 
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
 
16
 
 
17
"""Wire-level encoding and decoding of requests and responses for the smart
 
18
client and server.
 
19
"""
 
20
 
 
21
 
 
22
from cStringIO import StringIO
 
23
 
 
24
from bzrlib import errors
 
25
from bzrlib.smart import request
 
26
 
 
27
 
 
28
def _recv_tuple(from_file):
 
29
    req_line = from_file.readline()
 
30
    return _decode_tuple(req_line)
 
31
 
 
32
 
 
33
def _decode_tuple(req_line):
 
34
    if req_line == None or req_line == '':
 
35
        return None
 
36
    if req_line[-1] != '\n':
 
37
        raise errors.SmartProtocolError("request %r not terminated" % req_line)
 
38
    return tuple(req_line[:-1].split('\x01'))
 
39
 
 
40
 
 
41
def _encode_tuple(args):
 
42
    """Encode the tuple args to a bytestream."""
 
43
    return '\x01'.join(args) + '\n'
 
44
 
 
45
 
 
46
class SmartProtocolBase(object):
 
47
    """Methods common to client and server"""
 
48
 
 
49
    # TODO: this only actually accomodates a single block; possibly should
 
50
    # support multiple chunks?
 
51
    def _encode_bulk_data(self, body):
 
52
        """Encode body as a bulk data chunk."""
 
53
        return ''.join(('%d\n' % len(body), body, 'done\n'))
 
54
 
 
55
    def _serialise_offsets(self, offsets):
 
56
        """Serialise a readv offset list."""
 
57
        txt = []
 
58
        for start, length in offsets:
 
59
            txt.append('%d,%d' % (start, length))
 
60
        return '\n'.join(txt)
 
61
        
 
62
 
 
63
class SmartServerRequestProtocolOne(SmartProtocolBase):
 
64
    """Server-side encoding and decoding logic for smart version 1."""
 
65
    
 
66
    def __init__(self, backing_transport, write_func):
 
67
        self._backing_transport = backing_transport
 
68
        self.excess_buffer = ''
 
69
        self._finished = False
 
70
        self.in_buffer = ''
 
71
        self.has_dispatched = False
 
72
        self.request = None
 
73
        self._body_decoder = None
 
74
        self._write_func = write_func
 
75
 
 
76
    def accept_bytes(self, bytes):
 
77
        """Take bytes, and advance the internal state machine appropriately.
 
78
        
 
79
        :param bytes: must be a byte string
 
80
        """
 
81
        assert isinstance(bytes, str)
 
82
        self.in_buffer += bytes
 
83
        if not self.has_dispatched:
 
84
            if '\n' not in self.in_buffer:
 
85
                # no command line yet
 
86
                return
 
87
            self.has_dispatched = True
 
88
            try:
 
89
                first_line, self.in_buffer = self.in_buffer.split('\n', 1)
 
90
                first_line += '\n'
 
91
                req_args = _decode_tuple(first_line)
 
92
                self.request = request.SmartServerRequestHandler(
 
93
                    self._backing_transport, commands=request.request_handlers)
 
94
                self.request.dispatch_command(req_args[0], req_args[1:])
 
95
                if self.request.finished_reading:
 
96
                    # trivial request
 
97
                    self.excess_buffer = self.in_buffer
 
98
                    self.in_buffer = ''
 
99
                    self._send_response(self.request.response)
 
100
            except KeyboardInterrupt:
 
101
                raise
 
102
            except Exception, exception:
 
103
                # everything else: pass to client, flush, and quit
 
104
                self._send_response(request.FailedSmartServerResponse(
 
105
                    ('error', str(exception))))
 
106
                return
 
107
 
 
108
        if self.has_dispatched:
 
109
            if self._finished:
 
110
                # nothing to do.XXX: this routine should be a single state 
 
111
                # machine too.
 
112
                self.excess_buffer += self.in_buffer
 
113
                self.in_buffer = ''
 
114
                return
 
115
            if self._body_decoder is None:
 
116
                self._body_decoder = LengthPrefixedBodyDecoder()
 
117
            self._body_decoder.accept_bytes(self.in_buffer)
 
118
            self.in_buffer = self._body_decoder.unused_data
 
119
            body_data = self._body_decoder.read_pending_data()
 
120
            self.request.accept_body(body_data)
 
121
            if self._body_decoder.finished_reading:
 
122
                self.request.end_of_body()
 
123
                assert self.request.finished_reading, \
 
124
                    "no more body, request not finished"
 
125
            if self.request.response is not None:
 
126
                self._send_response(self.request.response)
 
127
                self.excess_buffer = self.in_buffer
 
128
                self.in_buffer = ''
 
129
            else:
 
130
                assert not self.request.finished_reading, \
 
131
                    "no response and we have finished reading."
 
132
 
 
133
    def _send_response(self, response):
 
134
        """Send a smart server response down the output stream."""
 
135
        assert not self._finished, 'response already sent'
 
136
        response.is_successful()
 
137
        args = response.args
 
138
        body = response.body
 
139
        self._finished = True
 
140
        self._write_protocol_version()
 
141
        self._write_func(_encode_tuple(args))
 
142
        if body is not None:
 
143
            assert isinstance(body, str), 'body must be a str'
 
144
            bytes = self._encode_bulk_data(body)
 
145
            self._write_func(bytes)
 
146
 
 
147
    def _write_protocol_version(self):
 
148
        """Write any prefixes this protocol requires.
 
149
        
 
150
        Version one doesn't send protocol versions.
 
151
        """
 
152
 
 
153
    def next_read_size(self):
 
154
        if self._finished:
 
155
            return 0
 
156
        if self._body_decoder is None:
 
157
            return 1
 
158
        else:
 
159
            return self._body_decoder.next_read_size()
 
160
 
 
161
 
 
162
class SmartServerRequestProtocolTwo(SmartServerRequestProtocolOne):
 
163
    r"""Version two of the server side of the smart protocol.
 
164
   
 
165
    This prefixes responses with the protocol version: "2\n".
 
166
    """
 
167
 
 
168
    def _write_protocol_version(self):
 
169
        r"""Write any prefixes this protocol requires.
 
170
        
 
171
        Version two sends "2\n".
 
172
        """
 
173
        self._write_func('2\n')
 
174
 
 
175
 
 
176
class LengthPrefixedBodyDecoder(object):
 
177
    """Decodes the length-prefixed bulk data."""
 
178
    
 
179
    def __init__(self):
 
180
        self.bytes_left = None
 
181
        self.finished_reading = False
 
182
        self.unused_data = ''
 
183
        self.state_accept = self._state_accept_expecting_length
 
184
        self.state_read = self._state_read_no_data
 
185
        self._in_buffer = ''
 
186
        self._trailer_buffer = ''
 
187
    
 
188
    def accept_bytes(self, bytes):
 
189
        """Decode as much of bytes as possible.
 
190
 
 
191
        If 'bytes' contains too much data it will be appended to
 
192
        self.unused_data.
 
193
 
 
194
        finished_reading will be set when no more data is required.  Further
 
195
        data will be appended to self.unused_data.
 
196
        """
 
197
        # accept_bytes is allowed to change the state
 
198
        current_state = self.state_accept
 
199
        self.state_accept(bytes)
 
200
        while current_state != self.state_accept:
 
201
            current_state = self.state_accept
 
202
            self.state_accept('')
 
203
 
 
204
    def next_read_size(self):
 
205
        if self.bytes_left is not None:
 
206
            # Ideally we want to read all the remainder of the body and the
 
207
            # trailer in one go.
 
208
            return self.bytes_left + 5
 
209
        elif self.state_accept == self._state_accept_reading_trailer:
 
210
            # Just the trailer left
 
211
            return 5 - len(self._trailer_buffer)
 
212
        elif self.state_accept == self._state_accept_expecting_length:
 
213
            # There's still at least 6 bytes left ('\n' to end the length, plus
 
214
            # 'done\n').
 
215
            return 6
 
216
        else:
 
217
            # Reading excess data.  Either way, 1 byte at a time is fine.
 
218
            return 1
 
219
        
 
220
    def read_pending_data(self):
 
221
        """Return any pending data that has been decoded."""
 
222
        return self.state_read()
 
223
 
 
224
    def _state_accept_expecting_length(self, bytes):
 
225
        self._in_buffer += bytes
 
226
        pos = self._in_buffer.find('\n')
 
227
        if pos == -1:
 
228
            return
 
229
        self.bytes_left = int(self._in_buffer[:pos])
 
230
        self._in_buffer = self._in_buffer[pos+1:]
 
231
        self.bytes_left -= len(self._in_buffer)
 
232
        self.state_accept = self._state_accept_reading_body
 
233
        self.state_read = self._state_read_in_buffer
 
234
 
 
235
    def _state_accept_reading_body(self, bytes):
 
236
        self._in_buffer += bytes
 
237
        self.bytes_left -= len(bytes)
 
238
        if self.bytes_left <= 0:
 
239
            # Finished with body
 
240
            if self.bytes_left != 0:
 
241
                self._trailer_buffer = self._in_buffer[self.bytes_left:]
 
242
                self._in_buffer = self._in_buffer[:self.bytes_left]
 
243
            self.bytes_left = None
 
244
            self.state_accept = self._state_accept_reading_trailer
 
245
        
 
246
    def _state_accept_reading_trailer(self, bytes):
 
247
        self._trailer_buffer += bytes
 
248
        # TODO: what if the trailer does not match "done\n"?  Should this raise
 
249
        # a ProtocolViolation exception?
 
250
        if self._trailer_buffer.startswith('done\n'):
 
251
            self.unused_data = self._trailer_buffer[len('done\n'):]
 
252
            self.state_accept = self._state_accept_reading_unused
 
253
            self.finished_reading = True
 
254
    
 
255
    def _state_accept_reading_unused(self, bytes):
 
256
        self.unused_data += bytes
 
257
 
 
258
    def _state_read_no_data(self):
 
259
        return ''
 
260
 
 
261
    def _state_read_in_buffer(self):
 
262
        result = self._in_buffer
 
263
        self._in_buffer = ''
 
264
        return result
 
265
 
 
266
 
 
267
class SmartClientRequestProtocolOne(SmartProtocolBase):
 
268
    """The client-side protocol for smart version 1."""
 
269
 
 
270
    def __init__(self, request):
 
271
        """Construct a SmartClientRequestProtocolOne.
 
272
 
 
273
        :param request: A SmartClientMediumRequest to serialise onto and
 
274
            deserialise from.
 
275
        """
 
276
        self._request = request
 
277
        self._body_buffer = None
 
278
 
 
279
    def call(self, *args):
 
280
        self._write_args(args)
 
281
        self._request.finished_writing()
 
282
 
 
283
    def call_with_body_bytes(self, args, body):
 
284
        """Make a remote call of args with body bytes 'body'.
 
285
 
 
286
        After calling this, call read_response_tuple to find the result out.
 
287
        """
 
288
        self._write_args(args)
 
289
        bytes = self._encode_bulk_data(body)
 
290
        self._request.accept_bytes(bytes)
 
291
        self._request.finished_writing()
 
292
 
 
293
    def call_with_body_readv_array(self, args, body):
 
294
        """Make a remote call with a readv array.
 
295
 
 
296
        The body is encoded with one line per readv offset pair. The numbers in
 
297
        each pair are separated by a comma, and no trailing \n is emitted.
 
298
        """
 
299
        self._write_args(args)
 
300
        readv_bytes = self._serialise_offsets(body)
 
301
        bytes = self._encode_bulk_data(readv_bytes)
 
302
        self._request.accept_bytes(bytes)
 
303
        self._request.finished_writing()
 
304
 
 
305
    def cancel_read_body(self):
 
306
        """After expecting a body, a response code may indicate one otherwise.
 
307
 
 
308
        This method lets the domain client inform the protocol that no body
 
309
        will be transmitted. This is a terminal method: after calling it the
 
310
        protocol is not able to be used further.
 
311
        """
 
312
        self._request.finished_reading()
 
313
 
 
314
    def read_response_tuple(self, expect_body=False):
 
315
        """Read a response tuple from the wire.
 
316
 
 
317
        This should only be called once.
 
318
        """
 
319
        result = self._recv_tuple()
 
320
        if not expect_body:
 
321
            self._request.finished_reading()
 
322
        return result
 
323
 
 
324
    def read_body_bytes(self, count=-1):
 
325
        """Read bytes from the body, decoding into a byte stream.
 
326
        
 
327
        We read all bytes at once to ensure we've checked the trailer for 
 
328
        errors, and then feed the buffer back as read_body_bytes is called.
 
329
        """
 
330
        if self._body_buffer is not None:
 
331
            return self._body_buffer.read(count)
 
332
        _body_decoder = LengthPrefixedBodyDecoder()
 
333
 
 
334
        while not _body_decoder.finished_reading:
 
335
            bytes_wanted = _body_decoder.next_read_size()
 
336
            bytes = self._request.read_bytes(bytes_wanted)
 
337
            _body_decoder.accept_bytes(bytes)
 
338
        self._request.finished_reading()
 
339
        self._body_buffer = StringIO(_body_decoder.read_pending_data())
 
340
        # XXX: TODO check the trailer result.
 
341
        return self._body_buffer.read(count)
 
342
 
 
343
    def _recv_tuple(self):
 
344
        """Receive a tuple from the medium request."""
 
345
        line = ''
 
346
        while not line or line[-1] != '\n':
 
347
            # TODO: this is inefficient - but tuples are short.
 
348
            new_char = self._request.read_bytes(1)
 
349
            line += new_char
 
350
            assert new_char != '', "end of file reading from server."
 
351
        return _decode_tuple(line)
 
352
 
 
353
    def query_version(self):
 
354
        """Return protocol version number of the server."""
 
355
        self.call('hello')
 
356
        resp = self.read_response_tuple()
 
357
        if resp == ('ok', '1'):
 
358
            return 1
 
359
        elif resp == ('ok', '2'):
 
360
            return 2
 
361
        else:
 
362
            raise errors.SmartProtocolError("bad response %r" % (resp,))
 
363
 
 
364
    def _write_args(self, args):
 
365
        self._write_protocol_version()
 
366
        bytes = _encode_tuple(args)
 
367
        self._request.accept_bytes(bytes)
 
368
 
 
369
    def _write_protocol_version(self):
 
370
        """Write any prefixes this protocol requires.
 
371
        
 
372
        Version one doesn't send protocol versions.
 
373
        """
 
374
 
 
375
 
 
376
class SmartClientRequestProtocolTwo(SmartClientRequestProtocolOne):
 
377
    r"""Version two of the client side of the smart protocol.
 
378
    
 
379
    This prefixes the request with the protocol version: "2\n".
 
380
    """
 
381
 
 
382
    _version_string = '2\n'
 
383
 
 
384
    def read_response_tuple(self, expect_body=False):
 
385
        """Read a response tuple from the wire.
 
386
 
 
387
        This should only be called once.
 
388
        """
 
389
        version = self._request.read_bytes(2)
 
390
        if version != SmartClientRequestProtocolTwo._version_string:
 
391
            raise errors.SmartProtocolError('bad protocol marker %r' % version)
 
392
        return SmartClientRequestProtocolOne.read_response_tuple(self, expect_body)
 
393
 
 
394
    def _write_protocol_version(self):
 
395
        r"""Write any prefixes this protocol requires.
 
396
        
 
397
        Version two sends "2\n".
 
398
        """
 
399
        self._request.accept_bytes(SmartClientRequestProtocolTwo._version_string)
 
400