/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

Merge from bzr.dev, resolving conflicts.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/protocol.py
 
1
# Copyright (C) 2006 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.args,
 
100
                        self.request.response.body)
 
101
            except KeyboardInterrupt:
 
102
                raise
 
103
            except Exception, exception:
 
104
                # everything else: pass to client, flush, and quit
 
105
                self._send_response(('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.args,
 
127
                    self.request.response.body)
 
128
                self.excess_buffer = self.in_buffer
 
129
                self.in_buffer = ''
 
130
            else:
 
131
                assert not self.request.finished_reading, \
 
132
                    "no response and we have finished reading."
 
133
 
 
134
    def _send_response(self, args, body=None):
 
135
        """Send a smart server response down the output stream."""
 
136
        assert not self._finished, 'response already sent'
 
137
        self._finished = True
 
138
        self._write_func(_encode_tuple(args))
 
139
        if body is not None:
 
140
            assert isinstance(body, str), 'body must be a str'
 
141
            bytes = self._encode_bulk_data(body)
 
142
            self._write_func(bytes)
 
143
 
 
144
    def next_read_size(self):
 
145
        if self._finished:
 
146
            return 0
 
147
        if self._body_decoder is None:
 
148
            return 1
 
149
        else:
 
150
            return self._body_decoder.next_read_size()
 
151
 
 
152
 
 
153
class LengthPrefixedBodyDecoder(object):
 
154
    """Decodes the length-prefixed bulk data."""
 
155
    
 
156
    def __init__(self):
 
157
        self.bytes_left = None
 
158
        self.finished_reading = False
 
159
        self.unused_data = ''
 
160
        self.state_accept = self._state_accept_expecting_length
 
161
        self.state_read = self._state_read_no_data
 
162
        self._in_buffer = ''
 
163
        self._trailer_buffer = ''
 
164
    
 
165
    def accept_bytes(self, bytes):
 
166
        """Decode as much of bytes as possible.
 
167
 
 
168
        If 'bytes' contains too much data it will be appended to
 
169
        self.unused_data.
 
170
 
 
171
        finished_reading will be set when no more data is required.  Further
 
172
        data will be appended to self.unused_data.
 
173
        """
 
174
        # accept_bytes is allowed to change the state
 
175
        current_state = self.state_accept
 
176
        self.state_accept(bytes)
 
177
        while current_state != self.state_accept:
 
178
            current_state = self.state_accept
 
179
            self.state_accept('')
 
180
 
 
181
    def next_read_size(self):
 
182
        if self.bytes_left is not None:
 
183
            # Ideally we want to read all the remainder of the body and the
 
184
            # trailer in one go.
 
185
            return self.bytes_left + 5
 
186
        elif self.state_accept == self._state_accept_reading_trailer:
 
187
            # Just the trailer left
 
188
            return 5 - len(self._trailer_buffer)
 
189
        elif self.state_accept == self._state_accept_expecting_length:
 
190
            # There's still at least 6 bytes left ('\n' to end the length, plus
 
191
            # 'done\n').
 
192
            return 6
 
193
        else:
 
194
            # Reading excess data.  Either way, 1 byte at a time is fine.
 
195
            return 1
 
196
        
 
197
    def read_pending_data(self):
 
198
        """Return any pending data that has been decoded."""
 
199
        return self.state_read()
 
200
 
 
201
    def _state_accept_expecting_length(self, bytes):
 
202
        self._in_buffer += bytes
 
203
        pos = self._in_buffer.find('\n')
 
204
        if pos == -1:
 
205
            return
 
206
        self.bytes_left = int(self._in_buffer[:pos])
 
207
        self._in_buffer = self._in_buffer[pos+1:]
 
208
        self.bytes_left -= len(self._in_buffer)
 
209
        self.state_accept = self._state_accept_reading_body
 
210
        self.state_read = self._state_read_in_buffer
 
211
 
 
212
    def _state_accept_reading_body(self, bytes):
 
213
        self._in_buffer += bytes
 
214
        self.bytes_left -= len(bytes)
 
215
        if self.bytes_left <= 0:
 
216
            # Finished with body
 
217
            if self.bytes_left != 0:
 
218
                self._trailer_buffer = self._in_buffer[self.bytes_left:]
 
219
                self._in_buffer = self._in_buffer[:self.bytes_left]
 
220
            self.bytes_left = None
 
221
            self.state_accept = self._state_accept_reading_trailer
 
222
        
 
223
    def _state_accept_reading_trailer(self, bytes):
 
224
        self._trailer_buffer += bytes
 
225
        # TODO: what if the trailer does not match "done\n"?  Should this raise
 
226
        # a ProtocolViolation exception?
 
227
        if self._trailer_buffer.startswith('done\n'):
 
228
            self.unused_data = self._trailer_buffer[len('done\n'):]
 
229
            self.state_accept = self._state_accept_reading_unused
 
230
            self.finished_reading = True
 
231
    
 
232
    def _state_accept_reading_unused(self, bytes):
 
233
        self.unused_data += bytes
 
234
 
 
235
    def _state_read_no_data(self):
 
236
        return ''
 
237
 
 
238
    def _state_read_in_buffer(self):
 
239
        result = self._in_buffer
 
240
        self._in_buffer = ''
 
241
        return result
 
242
 
 
243
 
 
244
# XXX: TODO: Create a SmartServerRequestHandler which will take the responsibility
 
245
# for delivering the data for a request. This could be done with as the
 
246
# StreamServer, though that would create conflation between request and response
 
247
# which may be undesirable.
 
248
 
 
249
 
 
250
class SmartClientRequestProtocolOne(SmartProtocolBase):
 
251
    """The client-side protocol for smart version 1."""
 
252
 
 
253
    def __init__(self, request):
 
254
        """Construct a SmartClientRequestProtocolOne.
 
255
 
 
256
        :param request: A SmartClientMediumRequest to serialise onto and
 
257
            deserialise from.
 
258
        """
 
259
        self._request = request
 
260
        self._body_buffer = None
 
261
 
 
262
    def call(self, *args):
 
263
        bytes = _encode_tuple(args)
 
264
        self._request.accept_bytes(bytes)
 
265
        self._request.finished_writing()
 
266
 
 
267
    def call_with_body_bytes(self, args, body):
 
268
        """Make a remote call of args with body bytes 'body'.
 
269
 
 
270
        After calling this, call read_response_tuple to find the result out.
 
271
        """
 
272
        bytes = _encode_tuple(args)
 
273
        self._request.accept_bytes(bytes)
 
274
        bytes = self._encode_bulk_data(body)
 
275
        self._request.accept_bytes(bytes)
 
276
        self._request.finished_writing()
 
277
 
 
278
    def call_with_body_readv_array(self, args, body):
 
279
        """Make a remote call with a readv array.
 
280
 
 
281
        The body is encoded with one line per readv offset pair. The numbers in
 
282
        each pair are separated by a comma, and no trailing \n is emitted.
 
283
        """
 
284
        bytes = _encode_tuple(args)
 
285
        self._request.accept_bytes(bytes)
 
286
        readv_bytes = self._serialise_offsets(body)
 
287
        bytes = self._encode_bulk_data(readv_bytes)
 
288
        self._request.accept_bytes(bytes)
 
289
        self._request.finished_writing()
 
290
 
 
291
    def cancel_read_body(self):
 
292
        """After expecting a body, a response code may indicate one otherwise.
 
293
 
 
294
        This method lets the domain client inform the protocol that no body
 
295
        will be transmitted. This is a terminal method: after calling it the
 
296
        protocol is not able to be used further.
 
297
        """
 
298
        self._request.finished_reading()
 
299
 
 
300
    def read_response_tuple(self, expect_body=False):
 
301
        """Read a response tuple from the wire.
 
302
 
 
303
        This should only be called once.
 
304
        """
 
305
        result = self._recv_tuple()
 
306
        if not expect_body:
 
307
            self._request.finished_reading()
 
308
        return result
 
309
 
 
310
    def read_body_bytes(self, count=-1):
 
311
        """Read bytes from the body, decoding into a byte stream.
 
312
        
 
313
        We read all bytes at once to ensure we've checked the trailer for 
 
314
        errors, and then feed the buffer back as read_body_bytes is called.
 
315
        """
 
316
        if self._body_buffer is not None:
 
317
            return self._body_buffer.read(count)
 
318
        _body_decoder = LengthPrefixedBodyDecoder()
 
319
 
 
320
        while not _body_decoder.finished_reading:
 
321
            bytes_wanted = _body_decoder.next_read_size()
 
322
            bytes = self._request.read_bytes(bytes_wanted)
 
323
            _body_decoder.accept_bytes(bytes)
 
324
        self._request.finished_reading()
 
325
        self._body_buffer = StringIO(_body_decoder.read_pending_data())
 
326
        # XXX: TODO check the trailer result.
 
327
        return self._body_buffer.read(count)
 
328
 
 
329
    def _recv_tuple(self):
 
330
        """Receive a tuple from the medium request."""
 
331
        line = ''
 
332
        while not line or line[-1] != '\n':
 
333
            # TODO: this is inefficient - but tuples are short.
 
334
            new_char = self._request.read_bytes(1)
 
335
            line += new_char
 
336
            assert new_char != '', "end of file reading from server."
 
337
        return _decode_tuple(line)
 
338
 
 
339
    def query_version(self):
 
340
        """Return protocol version number of the server."""
 
341
        self.call('hello')
 
342
        resp = self.read_response_tuple()
 
343
        if resp == ('ok', '1'):
 
344
            return 1
 
345
        else:
 
346
            raise errors.SmartProtocolError("bad response %r" % (resp,))
 
347
 
 
348