/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/message.py

  • Committer: Andrew Bennetts
  • Date: 2009-01-27 05:04:43 UTC
  • mfrom: (3960 +trunk)
  • mto: This revision was merged to the branch mainline in revision 3981.
  • Revision ID: andrew.bennetts@canonical.com-20090127050443-3yw5hhk10ss23hzu
Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/message.py
 
1
# Copyright (C) 2008 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
import collections
 
18
from cStringIO import StringIO
 
19
 
 
20
from bzrlib import (
 
21
    debug,
 
22
    errors,
 
23
    )
 
24
from bzrlib.trace import mutter
 
25
 
 
26
 
 
27
class MessageHandler(object):
 
28
    """Base class for handling messages received via the smart protocol.
 
29
 
 
30
    As parts of a message are received, the corresponding PART_received method
 
31
    will be called.
 
32
    """
 
33
 
 
34
    def __init__(self):
 
35
        self.headers = None
 
36
 
 
37
    def headers_received(self, headers):
 
38
        """Called when message headers are received.
 
39
        
 
40
        This default implementation just stores them in self.headers.
 
41
        """
 
42
        self.headers = headers
 
43
 
 
44
    def byte_part_received(self, byte):
 
45
        """Called when a 'byte' part is received.
 
46
 
 
47
        Note that a 'byte' part is a message part consisting of exactly one
 
48
        byte.
 
49
        """
 
50
        raise NotImplementedError(self.byte_received)
 
51
 
 
52
    def bytes_part_received(self, bytes):
 
53
        """Called when a 'bytes' part is received.
 
54
 
 
55
        A 'bytes' message part can contain any number of bytes.  It should not
 
56
        be confused with a 'byte' part, which is always a single byte.
 
57
        """
 
58
        raise NotImplementedError(self.bytes_received)
 
59
 
 
60
    def structure_part_received(self, structure):
 
61
        """Called when a 'structure' part is received.
 
62
 
 
63
        :param structure: some structured data, which will be some combination
 
64
            of list, dict, int, and str objects.
 
65
        """
 
66
        raise NotImplementedError(self.bytes_received)
 
67
 
 
68
    def protocol_error(self, exception):
 
69
        """Called when there is a protocol decoding error.
 
70
        
 
71
        The default implementation just re-raises the exception.
 
72
        """
 
73
        raise
 
74
    
 
75
    def end_received(self):
 
76
        """Called when the end of the message is received."""
 
77
        # No-op by default.
 
78
        pass
 
79
 
 
80
 
 
81
class ConventionalRequestHandler(MessageHandler):
 
82
    """A message handler for "conventional" requests.
 
83
 
 
84
    "Conventional" is used in the sense described in
 
85
    doc/developers/network-protocol.txt: a simple message with arguments and an
 
86
    optional body.
 
87
 
 
88
    Possible states:
 
89
     * expecting args
 
90
     * expecting body (terminated by receiving a post-body status)
 
91
     * finished
 
92
    """
 
93
 
 
94
    def __init__(self, request_handler, responder):
 
95
        MessageHandler.__init__(self)
 
96
        self.request_handler = request_handler
 
97
        self.responder = responder
 
98
        self.expecting = 'args'
 
99
        self._should_finish_body = False
 
100
        self._response_sent = False
 
101
 
 
102
    def protocol_error(self, exception):
 
103
        if self.responder.response_sent:
 
104
            # We can only send one response to a request, no matter how many
 
105
            # errors happen while processing it.
 
106
            return
 
107
        self.responder.send_error(exception)
 
108
 
 
109
    def byte_part_received(self, byte):
 
110
        if self.expecting == 'body':
 
111
            if byte == 'S':
 
112
                # Success.  Nothing more to come except the end of message.
 
113
                self.expecting = 'end'
 
114
            elif byte == 'E':
 
115
                # Error.  Expect an error structure.
 
116
                self.expecting = 'error'
 
117
            else:
 
118
                raise errors.SmartProtocolError(
 
119
                    'Non-success status byte in request body: %r' % (byte,))
 
120
        else:
 
121
            raise errors.SmartProtocolError(
 
122
                'Unexpected message part: byte(%r)' % (byte,))
 
123
 
 
124
    def structure_part_received(self, structure):
 
125
        if self.expecting == 'args':
 
126
            self._args_received(structure)
 
127
        elif self.expecting == 'error':
 
128
            self._error_received(structure)
 
129
        else:
 
130
            raise errors.SmartProtocolError(
 
131
                'Unexpected message part: structure(%r)' % (structure,))
 
132
 
 
133
    def _args_received(self, args):
 
134
        self.expecting = 'body'
 
135
        self.request_handler.dispatch_command(args[0], args[1:])
 
136
        if self.request_handler.finished_reading:
 
137
            self._response_sent = True
 
138
            self.responder.send_response(self.request_handler.response)
 
139
            self.expecting = 'end'
 
140
 
 
141
    def _error_received(self, error_args):
 
142
        self.expecting = 'end'
 
143
        self.request_handler.post_body_error_received(error_args)
 
144
 
 
145
    def bytes_part_received(self, bytes):
 
146
        if self.expecting == 'body':
 
147
            self._should_finish_body = True
 
148
            self.request_handler.accept_body(bytes)
 
149
        else:
 
150
            raise errors.SmartProtocolError(
 
151
                'Unexpected message part: bytes(%r)' % (bytes,))
 
152
 
 
153
    def end_received(self):
 
154
        if self.expecting not in ['body', 'end']:
 
155
            raise errors.SmartProtocolError(
 
156
                'End of message received prematurely (while expecting %s)'
 
157
                % (self.expecting,))
 
158
        self.expecting = 'nothing'
 
159
        self.request_handler.end_received()
 
160
        if not self.request_handler.finished_reading:
 
161
            raise errors.SmartProtocolError(
 
162
                "Complete conventional request was received, but request "
 
163
                "handler has not finished reading.")
 
164
        if not self._response_sent:
 
165
            self.responder.send_response(self.request_handler.response)
 
166
 
 
167
 
 
168
class ResponseHandler(object):
 
169
    """Abstract base class for an object that handles a smart response."""
 
170
 
 
171
    def read_response_tuple(self, expect_body=False):
 
172
        """Reads and returns the response tuple for the current request.
 
173
        
 
174
        :keyword expect_body: a boolean indicating if a body is expected in the
 
175
            response.  Some protocol versions needs this information to know
 
176
            when a response is finished.  If False, read_body_bytes should
 
177
            *not* be called afterwards.  Defaults to False.
 
178
        :returns: tuple of response arguments.
 
179
        """
 
180
        raise NotImplementedError(self.read_response_tuple)
 
181
 
 
182
    def read_body_bytes(self, count=-1):
 
183
        """Read and return some bytes from the body.
 
184
 
 
185
        :param count: if specified, read up to this many bytes.  By default,
 
186
            reads the entire body.
 
187
        :returns: str of bytes from the response body.
 
188
        """
 
189
        raise NotImplementedError(self.read_body_bytes)
 
190
 
 
191
    def read_streamed_body(self):
 
192
        """Returns an iterable that reads and returns a series of body chunks.
 
193
        """
 
194
        raise NotImplementedError(self.read_streamed_body)
 
195
 
 
196
    def cancel_read_body(self):
 
197
        """Stop expecting a body for this response.
 
198
 
 
199
        If expect_body was passed to read_response_tuple, this cancels that
 
200
        expectation (and thus finishes reading the response, allowing a new
 
201
        request to be issued).  This is useful if a response turns out to be an
 
202
        error rather than a normal result with a body.
 
203
        """
 
204
        raise NotImplementedError(self.cancel_read_body)
 
205
 
 
206
 
 
207
class ConventionalResponseHandler(MessageHandler, ResponseHandler):
 
208
 
 
209
    def __init__(self):
 
210
        MessageHandler.__init__(self)
 
211
        self.status = None
 
212
        self.args = None
 
213
        self._bytes_parts = collections.deque()
 
214
        self._body_started = False
 
215
        self._body_stream_status = None
 
216
        self._body = None
 
217
        self._body_error_args = None
 
218
        self.finished_reading = False
 
219
 
 
220
    def setProtoAndMediumRequest(self, protocol_decoder, medium_request):
 
221
        self._protocol_decoder = protocol_decoder
 
222
        self._medium_request = medium_request
 
223
 
 
224
    def byte_part_received(self, byte):
 
225
        if byte not in ['E', 'S']:
 
226
            raise errors.SmartProtocolError(
 
227
                'Unknown response status: %r' % (byte,))
 
228
        if self._body_started:
 
229
            if self._body_stream_status is not None:
 
230
                raise errors.SmartProtocolError(
 
231
                    'Unexpected byte part received: %r' % (byte,))
 
232
            self._body_stream_status = byte
 
233
        else:
 
234
            if self.status is not None:
 
235
                raise errors.SmartProtocolError(
 
236
                    'Unexpected byte part received: %r' % (byte,))
 
237
            self.status = byte
 
238
 
 
239
    def bytes_part_received(self, bytes):
 
240
        self._body_started = True
 
241
        self._bytes_parts.append(bytes)
 
242
 
 
243
    def structure_part_received(self, structure):
 
244
        if type(structure) is not tuple:
 
245
            raise errors.SmartProtocolError(
 
246
                'Args structure is not a sequence: %r' % (structure,))
 
247
        if not self._body_started:
 
248
            if self.args is not None:
 
249
                raise errors.SmartProtocolError(
 
250
                    'Unexpected structure received: %r (already got %r)'
 
251
                    % (structure, self.args))
 
252
            self.args = structure
 
253
        else:
 
254
            if self._body_stream_status != 'E':
 
255
                raise errors.SmartProtocolError(
 
256
                    'Unexpected structure received after body: %r'
 
257
                    % (structure,))
 
258
            self._body_error_args = structure
 
259
 
 
260
    def _wait_for_response_args(self):
 
261
        while self.args is None and not self.finished_reading:
 
262
            self._read_more()
 
263
 
 
264
    def _wait_for_response_end(self):
 
265
        while not self.finished_reading:
 
266
            self._read_more()
 
267
 
 
268
    def _read_more(self):
 
269
        next_read_size = self._protocol_decoder.next_read_size()
 
270
        if next_read_size == 0:
 
271
            # a complete request has been read.
 
272
            self.finished_reading = True
 
273
            self._medium_request.finished_reading()
 
274
            return
 
275
        bytes = self._medium_request.read_bytes(next_read_size)
 
276
        if bytes == '':
 
277
            # end of file encountered reading from server
 
278
            if 'hpss' in debug.debug_flags:
 
279
                mutter(
 
280
                    'decoder state: buf[:10]=%r, state_accept=%s',
 
281
                    self._protocol_decoder._get_in_buffer()[:10],
 
282
                    self._protocol_decoder.state_accept.__name__)
 
283
            raise errors.ConnectionReset(
 
284
                "please check connectivity and permissions",
 
285
                "(and try -Dhpss if further diagnosis is required)")
 
286
        self._protocol_decoder.accept_bytes(bytes)
 
287
 
 
288
    def protocol_error(self, exception):
 
289
        # Whatever the error is, we're done with this request.
 
290
        self.finished_reading = True
 
291
        self._medium_request.finished_reading()
 
292
        raise
 
293
        
 
294
    def read_response_tuple(self, expect_body=False):
 
295
        """Read a response tuple from the wire."""
 
296
        self._wait_for_response_args()
 
297
        if not expect_body:
 
298
            self._wait_for_response_end()
 
299
        if 'hpss' in debug.debug_flags:
 
300
            mutter('   result:   %r', self.args)
 
301
        if self.status == 'E':
 
302
            self._wait_for_response_end()
 
303
            _translate_error(self.args)
 
304
        return tuple(self.args)
 
305
 
 
306
    def read_body_bytes(self, count=-1):
 
307
        """Read bytes from the body, decoding into a byte stream.
 
308
        
 
309
        We read all bytes at once to ensure we've checked the trailer for 
 
310
        errors, and then feed the buffer back as read_body_bytes is called.
 
311
 
 
312
        Like the builtin file.read in Python, a count of -1 (the default) means
 
313
        read the entire body.
 
314
        """
 
315
        # TODO: we don't necessarily need to buffer the full request if count
 
316
        # != -1.  (2008/04/30, Andrew Bennetts)
 
317
        if self._body is None:
 
318
            self._wait_for_response_end()
 
319
            body_bytes = ''.join(self._bytes_parts)
 
320
            if 'hpss' in debug.debug_flags:
 
321
                mutter('              %d body bytes read', len(body_bytes))
 
322
            self._body = StringIO(body_bytes)
 
323
            self._bytes_parts = None
 
324
        return self._body.read(count)
 
325
 
 
326
    def read_streamed_body(self):
 
327
        while not self.finished_reading:
 
328
            while self._bytes_parts:
 
329
                bytes_part = self._bytes_parts.popleft()
 
330
                if 'hpss' in debug.debug_flags:
 
331
                    mutter('              %d byte part read', len(bytes_part))
 
332
                yield bytes_part
 
333
            self._read_more()
 
334
        if self._body_stream_status == 'E':
 
335
            _translate_error(self._body_error_args)
 
336
 
 
337
    def cancel_read_body(self):
 
338
        self._wait_for_response_end()
 
339
 
 
340
 
 
341
def _translate_error(error_tuple):
 
342
    # Many exceptions need some state from the requestor to be properly
 
343
    # translated (e.g. they need a branch object).  So this only translates a
 
344
    # few errors, and the rest are turned into a generic ErrorFromSmartServer.
 
345
    error_name = error_tuple[0]
 
346
    error_args = error_tuple[1:]
 
347
    if error_name == 'UnknownMethod':
 
348
        raise errors.UnknownSmartMethod(error_args[0])
 
349
    if error_name == 'LockContention':
 
350
        raise errors.LockContention('(remote lock)')
 
351
    elif error_name == 'LockFailed':
 
352
        raise errors.LockFailed(*error_args[:2])
 
353
    else:
 
354
        raise errors.ErrorFromSmartServer(error_tuple)