← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/protocol.py

  • Committer: Danny van Heumen
  • Date: 2010-03-09 21:42:11 UTC
  • mto: (4634.139.5 2.0)
  • mto: This revision was merged to the branch mainline in revision 5160.
  • Revision ID: danny@dannyvanheumen.nl-20100309214211-iqh42x6qcikgd9p3
Reverted now-useless TODO list.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/protocol.py
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
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
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""Wire-level encoding and decoding of requests and responses for the smart
18
18
client and server.
19
19
"""
20
20
 
21
 
 
 
21
import collections
22
22
from cStringIO import StringIO
 
23
import struct
 
24
import sys
 
25
import time
23
26
 
 
27
import bzrlib
 
28
from bzrlib import debug
24
29
from bzrlib import errors
25
 
from bzrlib.smart import request
 
30
from bzrlib.smart import message, request
 
31
from bzrlib.trace import log_exception_quietly, mutter
 
32
from bzrlib.bencode import bdecode_as_tuple, bencode
26
33
 
27
34
 
28
35
# Protocol version strings.  These are sent as prefixes of bzr requests and
31
38
REQUEST_VERSION_TWO = 'bzr request 2\n'
32
39
RESPONSE_VERSION_TWO = 'bzr response 2\n'
33
40
 
 
41
MESSAGE_VERSION_THREE = 'bzr message 3 (bzr 1.6)\n'
 
42
RESPONSE_VERSION_THREE = REQUEST_VERSION_THREE = MESSAGE_VERSION_THREE
 
43
 
34
44
 
35
45
def _recv_tuple(from_file):
36
46
    req_line = from_file.readline()
38
48
 
39
49
 
40
50
def _decode_tuple(req_line):
41
 
    if req_line == None or req_line == '':
 
51
    if req_line is None or req_line == '':
42
52
        return None
43
53
    if req_line[-1] != '\n':
44
54
        raise errors.SmartProtocolError("request %r not terminated" % req_line)
50
60
    return '\x01'.join(args) + '\n'
51
61
 
52
62
 
 
63
class Requester(object):
 
64
    """Abstract base class for an object that can issue requests on a smart
 
65
    medium.
 
66
    """
 
67
 
 
68
    def call(self, *args):
 
69
        """Make a remote call.
 
70
 
 
71
        :param args: the arguments of this call.
 
72
        """
 
73
        raise NotImplementedError(self.call)
 
74
 
 
75
    def call_with_body_bytes(self, args, body):
 
76
        """Make a remote call with a body.
 
77
 
 
78
        :param args: the arguments of this call.
 
79
        :type body: str
 
80
        :param body: the body to send with the request.
 
81
        """
 
82
        raise NotImplementedError(self.call_with_body_bytes)
 
83
 
 
84
    def call_with_body_readv_array(self, args, body):
 
85
        """Make a remote call with a readv array.
 
86
 
 
87
        :param args: the arguments of this call.
 
88
        :type body: iterable of (start, length) tuples.
 
89
        :param body: the readv ranges to send with this request.
 
90
        """
 
91
        raise NotImplementedError(self.call_with_body_readv_array)
 
92
 
 
93
    def set_headers(self, headers):
 
94
        raise NotImplementedError(self.set_headers)
 
95
 
 
96
 
53
97
class SmartProtocolBase(object):
54
98
    """Methods common to client and server"""
55
99
 
65
109
        for start, length in offsets:
66
110
            txt.append('%d,%d' % (start, length))
67
111
        return '\n'.join(txt)
68
 
        
 
112
 
69
113
 
70
114
class SmartServerRequestProtocolOne(SmartProtocolBase):
71
115
    """Server-side encoding and decoding logic for smart version 1."""
72
 
    
73
 
    def __init__(self, backing_transport, write_func):
 
116
 
 
117
    def __init__(self, backing_transport, write_func, root_client_path='/'):
74
118
        self._backing_transport = backing_transport
75
 
        self.excess_buffer = ''
 
119
        self._root_client_path = root_client_path
 
120
        self.unused_data = ''
76
121
        self._finished = False
77
122
        self.in_buffer = ''
78
 
        self.has_dispatched = False
 
123
        self._has_dispatched = False
79
124
        self.request = None
80
125
        self._body_decoder = None
81
126
        self._write_func = write_func
82
127
 
83
128
    def accept_bytes(self, bytes):
84
129
        """Take bytes, and advance the internal state machine appropriately.
85
 
        
 
130
 
86
131
        :param bytes: must be a byte string
87
132
        """
88
 
        assert isinstance(bytes, str)
 
133
        if not isinstance(bytes, str):
 
134
            raise ValueError(bytes)
89
135
        self.in_buffer += bytes
90
 
        if not self.has_dispatched:
 
136
        if not self._has_dispatched:
91
137
            if '\n' not in self.in_buffer:
92
138
                # no command line yet
93
139
                return
94
 
            self.has_dispatched = True
 
140
            self._has_dispatched = True
95
141
            try:
96
142
                first_line, self.in_buffer = self.in_buffer.split('\n', 1)
97
143
                first_line += '\n'
98
144
                req_args = _decode_tuple(first_line)
99
145
                self.request = request.SmartServerRequestHandler(
100
 
                    self._backing_transport, commands=request.request_handlers)
 
146
                    self._backing_transport, commands=request.request_handlers,
 
147
                    root_client_path=self._root_client_path)
101
148
                self.request.dispatch_command(req_args[0], req_args[1:])
102
149
                if self.request.finished_reading:
103
150
                    # trivial request
104
 
                    self.excess_buffer = self.in_buffer
 
151
                    self.unused_data = self.in_buffer
105
152
                    self.in_buffer = ''
106
153
                    self._send_response(self.request.response)
107
154
            except KeyboardInterrupt:
108
155
                raise
 
156
            except errors.UnknownSmartMethod, err:
 
157
                protocol_error = errors.SmartProtocolError(
 
158
                    "bad request %r" % (err.verb,))
 
159
                failure = request.FailedSmartServerResponse(
 
160
                    ('error', str(protocol_error)))
 
161
                self._send_response(failure)
 
162
                return
109
163
            except Exception, exception:
110
164
                # everything else: pass to client, flush, and quit
 
165
                log_exception_quietly()
111
166
                self._send_response(request.FailedSmartServerResponse(
112
167
                    ('error', str(exception))))
113
168
                return
114
169
 
115
 
        if self.has_dispatched:
 
170
        if self._has_dispatched:
116
171
            if self._finished:
117
 
                # nothing to do.XXX: this routine should be a single state 
 
172
                # nothing to do.XXX: this routine should be a single state
118
173
                # machine too.
119
 
                self.excess_buffer += self.in_buffer
 
174
                self.unused_data += self.in_buffer
120
175
                self.in_buffer = ''
121
176
                return
122
177
            if self._body_decoder is None:
127
182
            self.request.accept_body(body_data)
128
183
            if self._body_decoder.finished_reading:
129
184
                self.request.end_of_body()
130
 
                assert self.request.finished_reading, \
131
 
                    "no more body, request not finished"
 
185
                if not self.request.finished_reading:
 
186
                    raise AssertionError("no more body, request not finished")
132
187
            if self.request.response is not None:
133
188
                self._send_response(self.request.response)
134
 
                self.excess_buffer = self.in_buffer
 
189
                self.unused_data = self.in_buffer
135
190
                self.in_buffer = ''
136
191
            else:
137
 
                assert not self.request.finished_reading, \
138
 
                    "no response and we have finished reading."
 
192
                if self.request.finished_reading:
 
193
                    raise AssertionError(
 
194
                        "no response and we have finished reading.")
139
195
 
140
196
    def _send_response(self, response):
141
197
        """Send a smart server response down the output stream."""
142
 
        assert not self._finished, 'response already sent'
 
198
        if self._finished:
 
199
            raise AssertionError('response already sent')
143
200
        args = response.args
144
201
        body = response.body
145
202
        self._finished = True
147
204
        self._write_success_or_failure_prefix(response)
148
205
        self._write_func(_encode_tuple(args))
149
206
        if body is not None:
150
 
            assert isinstance(body, str), 'body must be a str'
 
207
            if not isinstance(body, str):
 
208
                raise ValueError(body)
151
209
            bytes = self._encode_bulk_data(body)
152
210
            self._write_func(bytes)
153
211
 
154
212
    def _write_protocol_version(self):
155
213
        """Write any prefixes this protocol requires.
156
 
        
 
214
 
157
215
        Version one doesn't send protocol versions.
158
216
        """
159
217
 
176
234
 
177
235
class SmartServerRequestProtocolTwo(SmartServerRequestProtocolOne):
178
236
    r"""Version two of the server side of the smart protocol.
179
 
   
 
237
 
180
238
    This prefixes responses with the value of RESPONSE_VERSION_TWO.
181
239
    """
182
240
 
 
241
    response_marker = RESPONSE_VERSION_TWO
 
242
    request_marker = REQUEST_VERSION_TWO
 
243
 
183
244
    def _write_success_or_failure_prefix(self, response):
184
245
        """Write the protocol specific success/failure prefix."""
185
246
        if response.is_successful():
189
250
 
190
251
    def _write_protocol_version(self):
191
252
        r"""Write any prefixes this protocol requires.
192
 
        
 
253
 
193
254
        Version two sends the value of RESPONSE_VERSION_TWO.
194
255
        """
195
 
        self._write_func(RESPONSE_VERSION_TWO)
196
 
 
197
 
 
198
 
class LengthPrefixedBodyDecoder(object):
199
 
    """Decodes the length-prefixed bulk data."""
200
 
    
 
256
        self._write_func(self.response_marker)
 
257
 
 
258
    def _send_response(self, response):
 
259
        """Send a smart server response down the output stream."""
 
260
        if (self._finished):
 
261
            raise AssertionError('response already sent')
 
262
        self._finished = True
 
263
        self._write_protocol_version()
 
264
        self._write_success_or_failure_prefix(response)
 
265
        self._write_func(_encode_tuple(response.args))
 
266
        if response.body is not None:
 
267
            if not isinstance(response.body, str):
 
268
                raise AssertionError('body must be a str')
 
269
            if not (response.body_stream is None):
 
270
                raise AssertionError(
 
271
                    'body_stream and body cannot both be set')
 
272
            bytes = self._encode_bulk_data(response.body)
 
273
            self._write_func(bytes)
 
274
        elif response.body_stream is not None:
 
275
            _send_stream(response.body_stream, self._write_func)
 
276
 
 
277
 
 
278
def _send_stream(stream, write_func):
 
279
    write_func('chunked\n')
 
280
    _send_chunks(stream, write_func)
 
281
    write_func('END\n')
 
282
 
 
283
 
 
284
def _send_chunks(stream, write_func):
 
285
    for chunk in stream:
 
286
        if isinstance(chunk, str):
 
287
            bytes = "%x\n%s" % (len(chunk), chunk)
 
288
            write_func(bytes)
 
289
        elif isinstance(chunk, request.FailedSmartServerResponse):
 
290
            write_func('ERR\n')
 
291
            _send_chunks(chunk.args, write_func)
 
292
            return
 
293
        else:
 
294
            raise errors.BzrError(
 
295
                'Chunks must be str or FailedSmartServerResponse, got %r'
 
296
                % chunk)
 
297
 
 
298
 
 
299
class _NeedMoreBytes(Exception):
 
300
    """Raise this inside a _StatefulDecoder to stop decoding until more bytes
 
301
    have been received.
 
302
    """
 
303
 
 
304
    def __init__(self, count=None):
 
305
        """Constructor.
 
306
 
 
307
        :param count: the total number of bytes needed by the current state.
 
308
            May be None if the number of bytes needed is unknown.
 
309
        """
 
310
        self.count = count
 
311
 
 
312
 
 
313
class _StatefulDecoder(object):
 
314
    """Base class for writing state machines to decode byte streams.
 
315
 
 
316
    Subclasses should provide a self.state_accept attribute that accepts bytes
 
317
    and, if appropriate, updates self.state_accept to a different function.
 
318
    accept_bytes will call state_accept as often as necessary to make sure the
 
319
    state machine has progressed as far as possible before it returns.
 
320
 
 
321
    See ProtocolThreeDecoder for an example subclass.
 
322
    """
 
323
 
201
324
    def __init__(self):
 
325
        self.finished_reading = False
 
326
        self._in_buffer_list = []
 
327
        self._in_buffer_len = 0
 
328
        self.unused_data = ''
202
329
        self.bytes_left = None
203
 
        self.finished_reading = False
204
 
        self.unused_data = ''
205
 
        self.state_accept = self._state_accept_expecting_length
206
 
        self.state_read = self._state_read_no_data
207
 
        self._in_buffer = ''
208
 
        self._trailer_buffer = ''
209
 
    
 
330
        self._number_needed_bytes = None
 
331
 
 
332
    def _get_in_buffer(self):
 
333
        if len(self._in_buffer_list) == 1:
 
334
            return self._in_buffer_list[0]
 
335
        in_buffer = ''.join(self._in_buffer_list)
 
336
        if len(in_buffer) != self._in_buffer_len:
 
337
            raise AssertionError(
 
338
                "Length of buffer did not match expected value: %s != %s"
 
339
                % self._in_buffer_len, len(in_buffer))
 
340
        self._in_buffer_list = [in_buffer]
 
341
        return in_buffer
 
342
 
 
343
    def _get_in_bytes(self, count):
 
344
        """Grab X bytes from the input_buffer.
 
345
 
 
346
        Callers should have already checked that self._in_buffer_len is >
 
347
        count. Note, this does not consume the bytes from the buffer. The
 
348
        caller will still need to call _get_in_buffer() and then
 
349
        _set_in_buffer() if they actually need to consume the bytes.
 
350
        """
 
351
        # check if we can yield the bytes from just the first entry in our list
 
352
        if len(self._in_buffer_list) == 0:
 
353
            raise AssertionError('Callers must be sure we have buffered bytes'
 
354
                ' before calling _get_in_bytes')
 
355
        if len(self._in_buffer_list[0]) > count:
 
356
            return self._in_buffer_list[0][:count]
 
357
        # We can't yield it from the first buffer, so collapse all buffers, and
 
358
        # yield it from that
 
359
        in_buf = self._get_in_buffer()
 
360
        return in_buf[:count]
 
361
 
 
362
    def _set_in_buffer(self, new_buf):
 
363
        if new_buf is not None:
 
364
            self._in_buffer_list = [new_buf]
 
365
            self._in_buffer_len = len(new_buf)
 
366
        else:
 
367
            self._in_buffer_list = []
 
368
            self._in_buffer_len = 0
 
369
 
210
370
    def accept_bytes(self, bytes):
211
371
        """Decode as much of bytes as possible.
212
372
 
217
377
        data will be appended to self.unused_data.
218
378
        """
219
379
        # accept_bytes is allowed to change the state
220
 
        current_state = self.state_accept
221
 
        self.state_accept(bytes)
222
 
        while current_state != self.state_accept:
 
380
        self._number_needed_bytes = None
 
381
        # lsprof puts a very large amount of time on this specific call for
 
382
        # large readv arrays
 
383
        self._in_buffer_list.append(bytes)
 
384
        self._in_buffer_len += len(bytes)
 
385
        try:
 
386
            # Run the function for the current state.
223
387
            current_state = self.state_accept
224
 
            self.state_accept('')
 
388
            self.state_accept()
 
389
            while current_state != self.state_accept:
 
390
                # The current state has changed.  Run the function for the new
 
391
                # current state, so that it can:
 
392
                #   - decode any unconsumed bytes left in a buffer, and
 
393
                #   - signal how many more bytes are expected (via raising
 
394
                #     _NeedMoreBytes).
 
395
                current_state = self.state_accept
 
396
                self.state_accept()
 
397
        except _NeedMoreBytes, e:
 
398
            self._number_needed_bytes = e.count
 
399
 
 
400
 
 
401
class ChunkedBodyDecoder(_StatefulDecoder):
 
402
    """Decoder for chunked body data.
 
403
 
 
404
    This is very similar the HTTP's chunked encoding.  See the description of
 
405
    streamed body data in `doc/developers/network-protocol.txt` for details.
 
406
    """
 
407
 
 
408
    def __init__(self):
 
409
        _StatefulDecoder.__init__(self)
 
410
        self.state_accept = self._state_accept_expecting_header
 
411
        self.chunk_in_progress = None
 
412
        self.chunks = collections.deque()
 
413
        self.error = False
 
414
        self.error_in_progress = None
 
415
 
 
416
    def next_read_size(self):
 
417
        # Note: the shortest possible chunk is 2 bytes: '0\n', and the
 
418
        # end-of-body marker is 4 bytes: 'END\n'.
 
419
        if self.state_accept == self._state_accept_reading_chunk:
 
420
            # We're expecting more chunk content.  So we're expecting at least
 
421
            # the rest of this chunk plus an END chunk.
 
422
            return self.bytes_left + 4
 
423
        elif self.state_accept == self._state_accept_expecting_length:
 
424
            if self._in_buffer_len == 0:
 
425
                # We're expecting a chunk length.  There's at least two bytes
 
426
                # left: a digit plus '\n'.
 
427
                return 2
 
428
            else:
 
429
                # We're in the middle of reading a chunk length.  So there's at
 
430
                # least one byte left, the '\n' that terminates the length.
 
431
                return 1
 
432
        elif self.state_accept == self._state_accept_reading_unused:
 
433
            return 1
 
434
        elif self.state_accept == self._state_accept_expecting_header:
 
435
            return max(0, len('chunked\n') - self._in_buffer_len)
 
436
        else:
 
437
            raise AssertionError("Impossible state: %r" % (self.state_accept,))
 
438
 
 
439
    def read_next_chunk(self):
 
440
        try:
 
441
            return self.chunks.popleft()
 
442
        except IndexError:
 
443
            return None
 
444
 
 
445
    def _extract_line(self):
 
446
        in_buf = self._get_in_buffer()
 
447
        pos = in_buf.find('\n')
 
448
        if pos == -1:
 
449
            # We haven't read a complete line yet, so request more bytes before
 
450
            # we continue.
 
451
            raise _NeedMoreBytes(1)
 
452
        line = in_buf[:pos]
 
453
        # Trim the prefix (including '\n' delimiter) from the _in_buffer.
 
454
        self._set_in_buffer(in_buf[pos+1:])
 
455
        return line
 
456
 
 
457
    def _finished(self):
 
458
        self.unused_data = self._get_in_buffer()
 
459
        self._in_buffer_list = []
 
460
        self._in_buffer_len = 0
 
461
        self.state_accept = self._state_accept_reading_unused
 
462
        if self.error:
 
463
            error_args = tuple(self.error_in_progress)
 
464
            self.chunks.append(request.FailedSmartServerResponse(error_args))
 
465
            self.error_in_progress = None
 
466
        self.finished_reading = True
 
467
 
 
468
    def _state_accept_expecting_header(self):
 
469
        prefix = self._extract_line()
 
470
        if prefix == 'chunked':
 
471
            self.state_accept = self._state_accept_expecting_length
 
472
        else:
 
473
            raise errors.SmartProtocolError(
 
474
                'Bad chunked body header: "%s"' % (prefix,))
 
475
 
 
476
    def _state_accept_expecting_length(self):
 
477
        prefix = self._extract_line()
 
478
        if prefix == 'ERR':
 
479
            self.error = True
 
480
            self.error_in_progress = []
 
481
            self._state_accept_expecting_length()
 
482
            return
 
483
        elif prefix == 'END':
 
484
            # We've read the end-of-body marker.
 
485
            # Any further bytes are unused data, including the bytes left in
 
486
            # the _in_buffer.
 
487
            self._finished()
 
488
            return
 
489
        else:
 
490
            self.bytes_left = int(prefix, 16)
 
491
            self.chunk_in_progress = ''
 
492
            self.state_accept = self._state_accept_reading_chunk
 
493
 
 
494
    def _state_accept_reading_chunk(self):
 
495
        in_buf = self._get_in_buffer()
 
496
        in_buffer_len = len(in_buf)
 
497
        self.chunk_in_progress += in_buf[:self.bytes_left]
 
498
        self._set_in_buffer(in_buf[self.bytes_left:])
 
499
        self.bytes_left -= in_buffer_len
 
500
        if self.bytes_left <= 0:
 
501
            # Finished with chunk
 
502
            self.bytes_left = None
 
503
            if self.error:
 
504
                self.error_in_progress.append(self.chunk_in_progress)
 
505
            else:
 
506
                self.chunks.append(self.chunk_in_progress)
 
507
            self.chunk_in_progress = None
 
508
            self.state_accept = self._state_accept_expecting_length
 
509
 
 
510
    def _state_accept_reading_unused(self):
 
511
        self.unused_data += self._get_in_buffer()
 
512
        self._in_buffer_list = []
 
513
 
 
514
 
 
515
class LengthPrefixedBodyDecoder(_StatefulDecoder):
 
516
    """Decodes the length-prefixed bulk data."""
 
517
 
 
518
    def __init__(self):
 
519
        _StatefulDecoder.__init__(self)
 
520
        self.state_accept = self._state_accept_expecting_length
 
521
        self.state_read = self._state_read_no_data
 
522
        self._body = ''
 
523
        self._trailer_buffer = ''
225
524
 
226
525
    def next_read_size(self):
227
526
        if self.bytes_left is not None:
238
537
        else:
239
538
            # Reading excess data.  Either way, 1 byte at a time is fine.
240
539
            return 1
241
 
        
 
540
 
242
541
    def read_pending_data(self):
243
542
        """Return any pending data that has been decoded."""
244
543
        return self.state_read()
245
544
 
246
 
    def _state_accept_expecting_length(self, bytes):
247
 
        self._in_buffer += bytes
248
 
        pos = self._in_buffer.find('\n')
 
545
    def _state_accept_expecting_length(self):
 
546
        in_buf = self._get_in_buffer()
 
547
        pos = in_buf.find('\n')
249
548
        if pos == -1:
250
549
            return
251
 
        self.bytes_left = int(self._in_buffer[:pos])
252
 
        self._in_buffer = self._in_buffer[pos+1:]
253
 
        self.bytes_left -= len(self._in_buffer)
 
550
        self.bytes_left = int(in_buf[:pos])
 
551
        self._set_in_buffer(in_buf[pos+1:])
254
552
        self.state_accept = self._state_accept_reading_body
255
 
        self.state_read = self._state_read_in_buffer
 
553
        self.state_read = self._state_read_body_buffer
256
554
 
257
 
    def _state_accept_reading_body(self, bytes):
258
 
        self._in_buffer += bytes
259
 
        self.bytes_left -= len(bytes)
 
555
    def _state_accept_reading_body(self):
 
556
        in_buf = self._get_in_buffer()
 
557
        self._body += in_buf
 
558
        self.bytes_left -= len(in_buf)
 
559
        self._set_in_buffer(None)
260
560
        if self.bytes_left <= 0:
261
561
            # Finished with body
262
562
            if self.bytes_left != 0:
263
 
                self._trailer_buffer = self._in_buffer[self.bytes_left:]
264
 
                self._in_buffer = self._in_buffer[:self.bytes_left]
 
563
                self._trailer_buffer = self._body[self.bytes_left:]
 
564
                self._body = self._body[:self.bytes_left]
265
565
            self.bytes_left = None
266
566
            self.state_accept = self._state_accept_reading_trailer
267
 
        
268
 
    def _state_accept_reading_trailer(self, bytes):
269
 
        self._trailer_buffer += bytes
 
567
 
 
568
    def _state_accept_reading_trailer(self):
 
569
        self._trailer_buffer += self._get_in_buffer()
 
570
        self._set_in_buffer(None)
270
571
        # TODO: what if the trailer does not match "done\n"?  Should this raise
271
572
        # a ProtocolViolation exception?
272
573
        if self._trailer_buffer.startswith('done\n'):
273
574
            self.unused_data = self._trailer_buffer[len('done\n'):]
274
575
            self.state_accept = self._state_accept_reading_unused
275
576
            self.finished_reading = True
276
 
    
277
 
    def _state_accept_reading_unused(self, bytes):
278
 
        self.unused_data += bytes
 
577
 
 
578
    def _state_accept_reading_unused(self):
 
579
        self.unused_data += self._get_in_buffer()
 
580
        self._set_in_buffer(None)
279
581
 
280
582
    def _state_read_no_data(self):
281
583
        return ''
282
584
 
283
 
    def _state_read_in_buffer(self):
284
 
        result = self._in_buffer
285
 
        self._in_buffer = ''
 
585
    def _state_read_body_buffer(self):
 
586
        result = self._body
 
587
        self._body = ''
286
588
        return result
287
589
 
288
590
 
289
 
class SmartClientRequestProtocolOne(SmartProtocolBase):
 
591
class SmartClientRequestProtocolOne(SmartProtocolBase, Requester,
 
592
                                    message.ResponseHandler):
290
593
    """The client-side protocol for smart version 1."""
291
594
 
292
595
    def __init__(self, request):
297
600
        """
298
601
        self._request = request
299
602
        self._body_buffer = None
 
603
        self._request_start_time = None
 
604
        self._last_verb = None
 
605
        self._headers = None
 
606
 
 
607
    def set_headers(self, headers):
 
608
        self._headers = dict(headers)
300
609
 
301
610
    def call(self, *args):
 
611
        if 'hpss' in debug.debug_flags:
 
612
            mutter('hpss call:   %s', repr(args)[1:-1])
 
613
            if getattr(self._request._medium, 'base', None) is not None:
 
614
                mutter('             (to %s)', self._request._medium.base)
 
615
            self._request_start_time = time.time()
302
616
        self._write_args(args)
303
617
        self._request.finished_writing()
 
618
        self._last_verb = args[0]
304
619
 
305
620
    def call_with_body_bytes(self, args, body):
306
621
        """Make a remote call of args with body bytes 'body'.
307
622
 
308
623
        After calling this, call read_response_tuple to find the result out.
309
624
        """
 
625
        if 'hpss' in debug.debug_flags:
 
626
            mutter('hpss call w/body: %s (%r...)', repr(args)[1:-1], body[:20])
 
627
            if getattr(self._request._medium, '_path', None) is not None:
 
628
                mutter('                  (to %s)', self._request._medium._path)
 
629
            mutter('              %d bytes', len(body))
 
630
            self._request_start_time = time.time()
 
631
            if 'hpssdetail' in debug.debug_flags:
 
632
                mutter('hpss body content: %s', body)
310
633
        self._write_args(args)
311
634
        bytes = self._encode_bulk_data(body)
312
635
        self._request.accept_bytes(bytes)
313
636
        self._request.finished_writing()
 
637
        self._last_verb = args[0]
314
638
 
315
639
    def call_with_body_readv_array(self, args, body):
316
640
        """Make a remote call with a readv array.
318
642
        The body is encoded with one line per readv offset pair. The numbers in
319
643
        each pair are separated by a comma, and no trailing \n is emitted.
320
644
        """
 
645
        if 'hpss' in debug.debug_flags:
 
646
            mutter('hpss call w/readv: %s', repr(args)[1:-1])
 
647
            if getattr(self._request._medium, '_path', None) is not None:
 
648
                mutter('                  (to %s)', self._request._medium._path)
 
649
            self._request_start_time = time.time()
321
650
        self._write_args(args)
322
651
        readv_bytes = self._serialise_offsets(body)
323
652
        bytes = self._encode_bulk_data(readv_bytes)
324
653
        self._request.accept_bytes(bytes)
325
654
        self._request.finished_writing()
 
655
        if 'hpss' in debug.debug_flags:
 
656
            mutter('              %d bytes in readv request', len(readv_bytes))
 
657
        self._last_verb = args[0]
 
658
 
 
659
    def call_with_body_stream(self, args, stream):
 
660
        # Protocols v1 and v2 don't support body streams.  So it's safe to
 
661
        # assume that a v1/v2 server doesn't support whatever method we're
 
662
        # trying to call with a body stream.
 
663
        self._request.finished_writing()
 
664
        self._request.finished_reading()
 
665
        raise errors.UnknownSmartMethod(args[0])
326
666
 
327
667
    def cancel_read_body(self):
328
668
        """After expecting a body, a response code may indicate one otherwise.
333
673
        """
334
674
        self._request.finished_reading()
335
675
 
336
 
    def read_response_tuple(self, expect_body=False):
337
 
        """Read a response tuple from the wire.
338
 
 
339
 
        This should only be called once.
340
 
        """
 
676
    def _read_response_tuple(self):
341
677
        result = self._recv_tuple()
 
678
        if 'hpss' in debug.debug_flags:
 
679
            if self._request_start_time is not None:
 
680
                mutter('   result:   %6.3fs  %s',
 
681
                       time.time() - self._request_start_time,
 
682
                       repr(result)[1:-1])
 
683
                self._request_start_time = None
 
684
            else:
 
685
                mutter('   result:   %s', repr(result)[1:-1])
 
686
        return result
 
687
 
 
688
    def read_response_tuple(self, expect_body=False):
 
689
        """Read a response tuple from the wire.
 
690
 
 
691
        This should only be called once.
 
692
        """
 
693
        result = self._read_response_tuple()
 
694
        self._response_is_unknown_method(result)
 
695
        self._raise_args_if_error(result)
342
696
        if not expect_body:
343
697
            self._request.finished_reading()
344
698
        return result
345
699
 
 
700
    def _raise_args_if_error(self, result_tuple):
 
701
        # Later protocol versions have an explicit flag in the protocol to say
 
702
        # if an error response is "failed" or not.  In version 1 we don't have
 
703
        # that luxury.  So here is a complete list of errors that can be
 
704
        # returned in response to existing version 1 smart requests.  Responses
 
705
        # starting with these codes are always "failed" responses.
 
706
        v1_error_codes = [
 
707
            'norepository',
 
708
            'NoSuchFile',
 
709
            'FileExists',
 
710
            'DirectoryNotEmpty',
 
711
            'ShortReadvError',
 
712
            'UnicodeEncodeError',
 
713
            'UnicodeDecodeError',
 
714
            'ReadOnlyError',
 
715
            'nobranch',
 
716
            'NoSuchRevision',
 
717
            'nosuchrevision',
 
718
            'LockContention',
 
719
            'UnlockableTransport',
 
720
            'LockFailed',
 
721
            'TokenMismatch',
 
722
            'ReadError',
 
723
            'PermissionDenied',
 
724
            ]
 
725
        if result_tuple[0] in v1_error_codes:
 
726
            self._request.finished_reading()
 
727
            raise errors.ErrorFromSmartServer(result_tuple)
 
728
 
 
729
    def _response_is_unknown_method(self, result_tuple):
 
730
        """Raise UnexpectedSmartServerResponse if the response is an 'unknonwn
 
731
        method' response to the request.
 
732
 
 
733
        :param response: The response from a smart client call_expecting_body
 
734
            call.
 
735
        :param verb: The verb used in that call.
 
736
        :raises: UnexpectedSmartServerResponse
 
737
        """
 
738
        if (result_tuple == ('error', "Generic bzr smart protocol error: "
 
739
                "bad request '%s'" % self._last_verb) or
 
740
              result_tuple == ('error', "Generic bzr smart protocol error: "
 
741
                "bad request u'%s'" % self._last_verb)):
 
742
            # The response will have no body, so we've finished reading.
 
743
            self._request.finished_reading()
 
744
            raise errors.UnknownSmartMethod(self._last_verb)
 
745
 
346
746
    def read_body_bytes(self, count=-1):
347
747
        """Read bytes from the body, decoding into a byte stream.
348
 
        
349
 
        We read all bytes at once to ensure we've checked the trailer for 
 
748
 
 
749
        We read all bytes at once to ensure we've checked the trailer for
350
750
        errors, and then feed the buffer back as read_body_bytes is called.
351
751
        """
352
752
        if self._body_buffer is not None:
354
754
        _body_decoder = LengthPrefixedBodyDecoder()
355
755
 
356
756
        while not _body_decoder.finished_reading:
357
 
            bytes_wanted = _body_decoder.next_read_size()
358
 
            bytes = self._request.read_bytes(bytes_wanted)
 
757
            bytes = self._request.read_bytes(_body_decoder.next_read_size())
 
758
            if bytes == '':
 
759
                # end of file encountered reading from server
 
760
                raise errors.ConnectionReset(
 
761
                    "Connection lost while reading response body.")
359
762
            _body_decoder.accept_bytes(bytes)
360
763
        self._request.finished_reading()
361
764
        self._body_buffer = StringIO(_body_decoder.read_pending_data())
362
765
        # XXX: TODO check the trailer result.
 
766
        if 'hpss' in debug.debug_flags:
 
767
            mutter('              %d body bytes read',
 
768
                   len(self._body_buffer.getvalue()))
363
769
        return self._body_buffer.read(count)
364
770
 
365
771
    def _recv_tuple(self):
366
772
        """Receive a tuple from the medium request."""
367
 
        return _decode_tuple(self._recv_line())
368
 
 
369
 
    def _recv_line(self):
370
 
        """Read an entire line from the medium request."""
371
 
        line = ''
372
 
        while not line or line[-1] != '\n':
373
 
            # TODO: this is inefficient - but tuples are short.
374
 
            new_char = self._request.read_bytes(1)
375
 
            line += new_char
376
 
            assert new_char != '', "end of file reading from server."
377
 
        return line
 
773
        return _decode_tuple(self._request.read_line())
378
774
 
379
775
    def query_version(self):
380
776
        """Return protocol version number of the server."""
394
790
 
395
791
    def _write_protocol_version(self):
396
792
        """Write any prefixes this protocol requires.
397
 
        
 
793
 
398
794
        Version one doesn't send protocol versions.
399
795
        """
400
796
 
401
797
 
402
798
class SmartClientRequestProtocolTwo(SmartClientRequestProtocolOne):
403
799
    """Version two of the client side of the smart protocol.
404
 
    
 
800
 
405
801
    This prefixes the request with the value of REQUEST_VERSION_TWO.
406
802
    """
407
803
 
 
804
    response_marker = RESPONSE_VERSION_TWO
 
805
    request_marker = REQUEST_VERSION_TWO
 
806
 
408
807
    def read_response_tuple(self, expect_body=False):
409
808
        """Read a response tuple from the wire.
410
809
 
411
810
        This should only be called once.
412
811
        """
413
812
        version = self._request.read_line()
414
 
        if version != RESPONSE_VERSION_TWO:
415
 
            raise errors.SmartProtocolError('bad protocol marker %r' % version)
416
 
        response_status = self._recv_line()
417
 
        if response_status not in ('success\n', 'failed\n'):
 
813
        if version != self.response_marker:
 
814
            self._request.finished_reading()
 
815
            raise errors.UnexpectedProtocolVersionMarker(version)
 
816
        response_status = self._request.read_line()
 
817
        result = SmartClientRequestProtocolOne._read_response_tuple(self)
 
818
        self._response_is_unknown_method(result)
 
819
        if response_status == 'success\n':
 
820
            self.response_status = True
 
821
            if not expect_body:
 
822
                self._request.finished_reading()
 
823
            return result
 
824
        elif response_status == 'failed\n':
 
825
            self.response_status = False
 
826
            self._request.finished_reading()
 
827
            raise errors.ErrorFromSmartServer(result)
 
828
        else:
418
829
            raise errors.SmartProtocolError(
419
830
                'bad protocol status %r' % response_status)
420
 
        self.response_status = response_status == 'success\n'
421
 
        return SmartClientRequestProtocolOne.read_response_tuple(self, expect_body)
422
831
 
423
832
    def _write_protocol_version(self):
424
 
        r"""Write any prefixes this protocol requires.
425
 
        
 
833
        """Write any prefixes this protocol requires.
 
834
 
426
835
        Version two sends the value of REQUEST_VERSION_TWO.
427
836
        """
428
 
        self._request.accept_bytes(REQUEST_VERSION_TWO)
 
837
        self._request.accept_bytes(self.request_marker)
 
838
 
 
839
    def read_streamed_body(self):
 
840
        """Read bytes from the body, decoding into a byte stream.
 
841
        """
 
842
        # Read no more than 64k at a time so that we don't risk error 10055 (no
 
843
        # buffer space available) on Windows.
 
844
        _body_decoder = ChunkedBodyDecoder()
 
845
        while not _body_decoder.finished_reading:
 
846
            bytes = self._request.read_bytes(_body_decoder.next_read_size())
 
847
            if bytes == '':
 
848
                # end of file encountered reading from server
 
849
                raise errors.ConnectionReset(
 
850
                    "Connection lost while reading streamed body.")
 
851
            _body_decoder.accept_bytes(bytes)
 
852
            for body_bytes in iter(_body_decoder.read_next_chunk, None):
 
853
                if 'hpss' in debug.debug_flags and type(body_bytes) is str:
 
854
                    mutter('              %d byte chunk read',
 
855
                           len(body_bytes))
 
856
                yield body_bytes
 
857
        self._request.finished_reading()
 
858
 
 
859
 
 
860
def build_server_protocol_three(backing_transport, write_func,
 
861
                                root_client_path):
 
862
    request_handler = request.SmartServerRequestHandler(
 
863
        backing_transport, commands=request.request_handlers,
 
864
        root_client_path=root_client_path)
 
865
    responder = ProtocolThreeResponder(write_func)
 
866
    message_handler = message.ConventionalRequestHandler(request_handler, responder)
 
867
    return ProtocolThreeDecoder(message_handler)
 
868
 
 
869
 
 
870
class ProtocolThreeDecoder(_StatefulDecoder):
 
871
 
 
872
    response_marker = RESPONSE_VERSION_THREE
 
873
    request_marker = REQUEST_VERSION_THREE
 
874
 
 
875
    def __init__(self, message_handler, expect_version_marker=False):
 
876
        _StatefulDecoder.__init__(self)
 
877
        self._has_dispatched = False
 
878
        # Initial state
 
879
        if expect_version_marker:
 
880
            self.state_accept = self._state_accept_expecting_protocol_version
 
881
            # We're expecting at least the protocol version marker + some
 
882
            # headers.
 
883
            self._number_needed_bytes = len(MESSAGE_VERSION_THREE) + 4
 
884
        else:
 
885
            self.state_accept = self._state_accept_expecting_headers
 
886
            self._number_needed_bytes = 4
 
887
        self.decoding_failed = False
 
888
        self.request_handler = self.message_handler = message_handler
 
889
 
 
890
    def accept_bytes(self, bytes):
 
891
        self._number_needed_bytes = None
 
892
        try:
 
893
            _StatefulDecoder.accept_bytes(self, bytes)
 
894
        except KeyboardInterrupt:
 
895
            raise
 
896
        except errors.SmartMessageHandlerError, exception:
 
897
            # We do *not* set self.decoding_failed here.  The message handler
 
898
            # has raised an error, but the decoder is still able to parse bytes
 
899
            # and determine when this message ends.
 
900
            if not isinstance(exception.exc_value, errors.UnknownSmartMethod):
 
901
                log_exception_quietly()
 
902
            self.message_handler.protocol_error(exception.exc_value)
 
903
            # The state machine is ready to continue decoding, but the
 
904
            # exception has interrupted the loop that runs the state machine.
 
905
            # So we call accept_bytes again to restart it.
 
906
            self.accept_bytes('')
 
907
        except Exception, exception:
 
908
            # The decoder itself has raised an exception.  We cannot continue
 
909
            # decoding.
 
910
            self.decoding_failed = True
 
911
            if isinstance(exception, errors.UnexpectedProtocolVersionMarker):
 
912
                # This happens during normal operation when the client tries a
 
913
                # protocol version the server doesn't understand, so no need to
 
914
                # log a traceback every time.
 
915
                # Note that this can only happen when
 
916
                # expect_version_marker=True, which is only the case on the
 
917
                # client side.
 
918
                pass
 
919
            else:
 
920
                log_exception_quietly()
 
921
            self.message_handler.protocol_error(exception)
 
922
 
 
923
    def _extract_length_prefixed_bytes(self):
 
924
        if self._in_buffer_len < 4:
 
925
            # A length prefix by itself is 4 bytes, and we don't even have that
 
926
            # many yet.
 
927
            raise _NeedMoreBytes(4)
 
928
        (length,) = struct.unpack('!L', self._get_in_bytes(4))
 
929
        end_of_bytes = 4 + length
 
930
        if self._in_buffer_len < end_of_bytes:
 
931
            # We haven't yet read as many bytes as the length-prefix says there
 
932
            # are.
 
933
            raise _NeedMoreBytes(end_of_bytes)
 
934
        # Extract the bytes from the buffer.
 
935
        in_buf = self._get_in_buffer()
 
936
        bytes = in_buf[4:end_of_bytes]
 
937
        self._set_in_buffer(in_buf[end_of_bytes:])
 
938
        return bytes
 
939
 
 
940
    def _extract_prefixed_bencoded_data(self):
 
941
        prefixed_bytes = self._extract_length_prefixed_bytes()
 
942
        try:
 
943
            decoded = bdecode_as_tuple(prefixed_bytes)
 
944
        except ValueError:
 
945
            raise errors.SmartProtocolError(
 
946
                'Bytes %r not bencoded' % (prefixed_bytes,))
 
947
        return decoded
 
948
 
 
949
    def _extract_single_byte(self):
 
950
        if self._in_buffer_len == 0:
 
951
            # The buffer is empty
 
952
            raise _NeedMoreBytes(1)
 
953
        in_buf = self._get_in_buffer()
 
954
        one_byte = in_buf[0]
 
955
        self._set_in_buffer(in_buf[1:])
 
956
        return one_byte
 
957
 
 
958
    def _state_accept_expecting_protocol_version(self):
 
959
        needed_bytes = len(MESSAGE_VERSION_THREE) - self._in_buffer_len
 
960
        in_buf = self._get_in_buffer()
 
961
        if needed_bytes > 0:
 
962
            # We don't have enough bytes to check if the protocol version
 
963
            # marker is right.  But we can check if it is already wrong by
 
964
            # checking that the start of MESSAGE_VERSION_THREE matches what
 
965
            # we've read so far.
 
966
            # [In fact, if the remote end isn't bzr we might never receive
 
967
            # len(MESSAGE_VERSION_THREE) bytes.  So if the bytes we have so far
 
968
            # are wrong then we should just raise immediately rather than
 
969
            # stall.]
 
970
            if not MESSAGE_VERSION_THREE.startswith(in_buf):
 
971
                # We have enough bytes to know the protocol version is wrong
 
972
                raise errors.UnexpectedProtocolVersionMarker(in_buf)
 
973
            raise _NeedMoreBytes(len(MESSAGE_VERSION_THREE))
 
974
        if not in_buf.startswith(MESSAGE_VERSION_THREE):
 
975
            raise errors.UnexpectedProtocolVersionMarker(in_buf)
 
976
        self._set_in_buffer(in_buf[len(MESSAGE_VERSION_THREE):])
 
977
        self.state_accept = self._state_accept_expecting_headers
 
978
 
 
979
    def _state_accept_expecting_headers(self):
 
980
        decoded = self._extract_prefixed_bencoded_data()
 
981
        if type(decoded) is not dict:
 
982
            raise errors.SmartProtocolError(
 
983
                'Header object %r is not a dict' % (decoded,))
 
984
        self.state_accept = self._state_accept_expecting_message_part
 
985
        try:
 
986
            self.message_handler.headers_received(decoded)
 
987
        except:
 
988
            raise errors.SmartMessageHandlerError(sys.exc_info())
 
989
 
 
990
    def _state_accept_expecting_message_part(self):
 
991
        message_part_kind = self._extract_single_byte()
 
992
        if message_part_kind == 'o':
 
993
            self.state_accept = self._state_accept_expecting_one_byte
 
994
        elif message_part_kind == 's':
 
995
            self.state_accept = self._state_accept_expecting_structure
 
996
        elif message_part_kind == 'b':
 
997
            self.state_accept = self._state_accept_expecting_bytes
 
998
        elif message_part_kind == 'e':
 
999
            self.done()
 
1000
        else:
 
1001
            raise errors.SmartProtocolError(
 
1002
                'Bad message kind byte: %r' % (message_part_kind,))
 
1003
 
 
1004
    def _state_accept_expecting_one_byte(self):
 
1005
        byte = self._extract_single_byte()
 
1006
        self.state_accept = self._state_accept_expecting_message_part
 
1007
        try:
 
1008
            self.message_handler.byte_part_received(byte)
 
1009
        except:
 
1010
            raise errors.SmartMessageHandlerError(sys.exc_info())
 
1011
 
 
1012
    def _state_accept_expecting_bytes(self):
 
1013
        # XXX: this should not buffer whole message part, but instead deliver
 
1014
        # the bytes as they arrive.
 
1015
        prefixed_bytes = self._extract_length_prefixed_bytes()
 
1016
        self.state_accept = self._state_accept_expecting_message_part
 
1017
        try:
 
1018
            self.message_handler.bytes_part_received(prefixed_bytes)
 
1019
        except:
 
1020
            raise errors.SmartMessageHandlerError(sys.exc_info())
 
1021
 
 
1022
    def _state_accept_expecting_structure(self):
 
1023
        structure = self._extract_prefixed_bencoded_data()
 
1024
        self.state_accept = self._state_accept_expecting_message_part
 
1025
        try:
 
1026
            self.message_handler.structure_part_received(structure)
 
1027
        except:
 
1028
            raise errors.SmartMessageHandlerError(sys.exc_info())
 
1029
 
 
1030
    def done(self):
 
1031
        self.unused_data = self._get_in_buffer()
 
1032
        self._set_in_buffer(None)
 
1033
        self.state_accept = self._state_accept_reading_unused
 
1034
        try:
 
1035
            self.message_handler.end_received()
 
1036
        except:
 
1037
            raise errors.SmartMessageHandlerError(sys.exc_info())
 
1038
 
 
1039
    def _state_accept_reading_unused(self):
 
1040
        self.unused_data += self._get_in_buffer()
 
1041
        self._set_in_buffer(None)
 
1042
 
 
1043
    def next_read_size(self):
 
1044
        if self.state_accept == self._state_accept_reading_unused:
 
1045
            return 0
 
1046
        elif self.decoding_failed:
 
1047
            # An exception occured while processing this message, probably from
 
1048
            # self.message_handler.  We're not sure that this state machine is
 
1049
            # in a consistent state, so just signal that we're done (i.e. give
 
1050
            # up).
 
1051
            return 0
 
1052
        else:
 
1053
            if self._number_needed_bytes is not None:
 
1054
                return self._number_needed_bytes - self._in_buffer_len
 
1055
            else:
 
1056
                raise AssertionError("don't know how many bytes are expected!")
 
1057
 
 
1058
 
 
1059
class _ProtocolThreeEncoder(object):
 
1060
 
 
1061
    response_marker = request_marker = MESSAGE_VERSION_THREE
 
1062
 
 
1063
    def __init__(self, write_func):
 
1064
        self._buf = []
 
1065
        self._real_write_func = write_func
 
1066
 
 
1067
    def _write_func(self, bytes):
 
1068
        self._buf.append(bytes)
 
1069
        if len(self._buf) > 100:
 
1070
            self.flush()
 
1071
 
 
1072
    def flush(self):
 
1073
        if self._buf:
 
1074
            self._real_write_func(''.join(self._buf))
 
1075
            del self._buf[:]
 
1076
 
 
1077
    def _serialise_offsets(self, offsets):
 
1078
        """Serialise a readv offset list."""
 
1079
        txt = []
 
1080
        for start, length in offsets:
 
1081
            txt.append('%d,%d' % (start, length))
 
1082
        return '\n'.join(txt)
 
1083
 
 
1084
    def _write_protocol_version(self):
 
1085
        self._write_func(MESSAGE_VERSION_THREE)
 
1086
 
 
1087
    def _write_prefixed_bencode(self, structure):
 
1088
        bytes = bencode(structure)
 
1089
        self._write_func(struct.pack('!L', len(bytes)))
 
1090
        self._write_func(bytes)
 
1091
 
 
1092
    def _write_headers(self, headers):
 
1093
        self._write_prefixed_bencode(headers)
 
1094
 
 
1095
    def _write_structure(self, args):
 
1096
        self._write_func('s')
 
1097
        utf8_args = []
 
1098
        for arg in args:
 
1099
            if type(arg) is unicode:
 
1100
                utf8_args.append(arg.encode('utf8'))
 
1101
            else:
 
1102
                utf8_args.append(arg)
 
1103
        self._write_prefixed_bencode(utf8_args)
 
1104
 
 
1105
    def _write_end(self):
 
1106
        self._write_func('e')
 
1107
        self.flush()
 
1108
 
 
1109
    def _write_prefixed_body(self, bytes):
 
1110
        self._write_func('b')
 
1111
        self._write_func(struct.pack('!L', len(bytes)))
 
1112
        self._write_func(bytes)
 
1113
 
 
1114
    def _write_chunked_body_start(self):
 
1115
        self._write_func('oC')
 
1116
 
 
1117
    def _write_error_status(self):
 
1118
        self._write_func('oE')
 
1119
 
 
1120
    def _write_success_status(self):
 
1121
        self._write_func('oS')
 
1122
 
 
1123
 
 
1124
class ProtocolThreeResponder(_ProtocolThreeEncoder):
 
1125
 
 
1126
    def __init__(self, write_func):
 
1127
        _ProtocolThreeEncoder.__init__(self, write_func)
 
1128
        self.response_sent = False
 
1129
        self._headers = {'Software version': bzrlib.__version__}
 
1130
 
 
1131
    def send_error(self, exception):
 
1132
        if self.response_sent:
 
1133
            raise AssertionError(
 
1134
                "send_error(%s) called, but response already sent."
 
1135
                % (exception,))
 
1136
        if isinstance(exception, errors.UnknownSmartMethod):
 
1137
            failure = request.FailedSmartServerResponse(
 
1138
                ('UnknownMethod', exception.verb))
 
1139
            self.send_response(failure)
 
1140
            return
 
1141
        self.response_sent = True
 
1142
        self._write_protocol_version()
 
1143
        self._write_headers(self._headers)
 
1144
        self._write_error_status()
 
1145
        self._write_structure(('error', str(exception)))
 
1146
        self._write_end()
 
1147
 
 
1148
    def send_response(self, response):
 
1149
        if self.response_sent:
 
1150
            raise AssertionError(
 
1151
                "send_response(%r) called, but response already sent."
 
1152
                % (response,))
 
1153
        self.response_sent = True
 
1154
        self._write_protocol_version()
 
1155
        self._write_headers(self._headers)
 
1156
        if response.is_successful():
 
1157
            self._write_success_status()
 
1158
        else:
 
1159
            self._write_error_status()
 
1160
        self._write_structure(response.args)
 
1161
        if response.body is not None:
 
1162
            self._write_prefixed_body(response.body)
 
1163
        elif response.body_stream is not None:
 
1164
            for exc_info, chunk in _iter_with_errors(response.body_stream):
 
1165
                if exc_info is not None:
 
1166
                    self._write_error_status()
 
1167
                    error_struct = request._translate_error(exc_info[1])
 
1168
                    self._write_structure(error_struct)
 
1169
                    break
 
1170
                else:
 
1171
                    if isinstance(chunk, request.FailedSmartServerResponse):
 
1172
                        self._write_error_status()
 
1173
                        self._write_structure(chunk.args)
 
1174
                        break
 
1175
                    self._write_prefixed_body(chunk)
 
1176
        self._write_end()
 
1177
 
 
1178
 
 
1179
def _iter_with_errors(iterable):
 
1180
    """Handle errors from iterable.next().
 
1181
 
 
1182
    Use like::
 
1183
 
 
1184
        for exc_info, value in _iter_with_errors(iterable):
 
1185
            ...
 
1186
 
 
1187
    This is a safer alternative to::
 
1188
 
 
1189
        try:
 
1190
            for value in iterable:
 
1191
               ...
 
1192
        except:
 
1193
            ...
 
1194
 
 
1195
    Because the latter will catch errors from the for-loop body, not just
 
1196
    iterable.next()
 
1197
 
 
1198
    If an error occurs, exc_info will be a exc_info tuple, and the generator
 
1199
    will terminate.  Otherwise exc_info will be None, and value will be the
 
1200
    value from iterable.next().  Note that KeyboardInterrupt and SystemExit
 
1201
    will not be itercepted.
 
1202
    """
 
1203
    iterator = iter(iterable)
 
1204
    while True:
 
1205
        try:
 
1206
            yield None, iterator.next()
 
1207
        except StopIteration:
 
1208
            return
 
1209
        except (KeyboardInterrupt, SystemExit):
 
1210
            raise
 
1211
        except Exception:
 
1212
            mutter('_iter_with_errors caught error')
 
1213
            log_exception_quietly()
 
1214
            yield sys.exc_info(), None
 
1215
            return
 
1216
 
 
1217
 
 
1218
class ProtocolThreeRequester(_ProtocolThreeEncoder, Requester):
 
1219
 
 
1220
    def __init__(self, medium_request):
 
1221
        _ProtocolThreeEncoder.__init__(self, medium_request.accept_bytes)
 
1222
        self._medium_request = medium_request
 
1223
        self._headers = {}
 
1224
 
 
1225
    def set_headers(self, headers):
 
1226
        self._headers = headers.copy()
 
1227
 
 
1228
    def call(self, *args):
 
1229
        if 'hpss' in debug.debug_flags:
 
1230
            mutter('hpss call:   %s', repr(args)[1:-1])
 
1231
            base = getattr(self._medium_request._medium, 'base', None)
 
1232
            if base is not None:
 
1233
                mutter('             (to %s)', base)
 
1234
            self._request_start_time = time.time()
 
1235
        self._write_protocol_version()
 
1236
        self._write_headers(self._headers)
 
1237
        self._write_structure(args)
 
1238
        self._write_end()
 
1239
        self._medium_request.finished_writing()
 
1240
 
 
1241
    def call_with_body_bytes(self, args, body):
 
1242
        """Make a remote call of args with body bytes 'body'.
 
1243
 
 
1244
        After calling this, call read_response_tuple to find the result out.
 
1245
        """
 
1246
        if 'hpss' in debug.debug_flags:
 
1247
            mutter('hpss call w/body: %s (%r...)', repr(args)[1:-1], body[:20])
 
1248
            path = getattr(self._medium_request._medium, '_path', None)
 
1249
            if path is not None:
 
1250
                mutter('                  (to %s)', path)
 
1251
            mutter('              %d bytes', len(body))
 
1252
            self._request_start_time = time.time()
 
1253
        self._write_protocol_version()
 
1254
        self._write_headers(self._headers)
 
1255
        self._write_structure(args)
 
1256
        self._write_prefixed_body(body)
 
1257
        self._write_end()
 
1258
        self._medium_request.finished_writing()
 
1259
 
 
1260
    def call_with_body_readv_array(self, args, body):
 
1261
        """Make a remote call with a readv array.
 
1262
 
 
1263
        The body is encoded with one line per readv offset pair. The numbers in
 
1264
        each pair are separated by a comma, and no trailing \n is emitted.
 
1265
        """
 
1266
        if 'hpss' in debug.debug_flags:
 
1267
            mutter('hpss call w/readv: %s', repr(args)[1:-1])
 
1268
            path = getattr(self._medium_request._medium, '_path', None)
 
1269
            if path is not None:
 
1270
                mutter('                  (to %s)', path)
 
1271
            self._request_start_time = time.time()
 
1272
        self._write_protocol_version()
 
1273
        self._write_headers(self._headers)
 
1274
        self._write_structure(args)
 
1275
        readv_bytes = self._serialise_offsets(body)
 
1276
        if 'hpss' in debug.debug_flags:
 
1277
            mutter('              %d bytes in readv request', len(readv_bytes))
 
1278
        self._write_prefixed_body(readv_bytes)
 
1279
        self._write_end()
 
1280
        self._medium_request.finished_writing()
 
1281
 
 
1282
    def call_with_body_stream(self, args, stream):
 
1283
        if 'hpss' in debug.debug_flags:
 
1284
            mutter('hpss call w/body stream: %r', args)
 
1285
            path = getattr(self._medium_request._medium, '_path', None)
 
1286
            if path is not None:
 
1287
                mutter('                  (to %s)', path)
 
1288
            self._request_start_time = time.time()
 
1289
        self._write_protocol_version()
 
1290
        self._write_headers(self._headers)
 
1291
        self._write_structure(args)
 
1292
        # TODO: notice if the server has sent an early error reply before we
 
1293
        #       have finished sending the stream.  We would notice at the end
 
1294
        #       anyway, but if the medium can deliver it early then it's good
 
1295
        #       to short-circuit the whole request...
 
1296
        for exc_info, part in _iter_with_errors(stream):
 
1297
            if exc_info is not None:
 
1298
                # Iterating the stream failed.  Cleanly abort the request.
 
1299
                self._write_error_status()
 
1300
                # Currently the client unconditionally sends ('error',) as the
 
1301
                # error args.
 
1302
                self._write_structure(('error',))
 
1303
                self._write_end()
 
1304
                self._medium_request.finished_writing()
 
1305
                raise exc_info[0], exc_info[1], exc_info[2]
 
1306
            else:
 
1307
                self._write_prefixed_body(part)
 
1308
                self.flush()
 
1309
        self._write_end()
 
1310
        self._medium_request.finished_writing()
429
1311