1
# Copyright (C) 2008 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
from __future__ import absolute_import
20
from collections.abc import deque
21
except ImportError: # python < 3.7
22
from collections import deque
28
from ...sixish import (
31
from ...trace import mutter
34
class MessageHandler(object):
35
"""Base class for handling messages received via the smart protocol.
37
As parts of a message are received, the corresponding PART_received method
44
def headers_received(self, headers):
45
"""Called when message headers are received.
47
This default implementation just stores them in self.headers.
49
self.headers = headers
51
def byte_part_received(self, byte):
52
"""Called when a 'byte' part is received.
54
Note that a 'byte' part is a message part consisting of exactly one
57
raise NotImplementedError(self.byte_received)
59
def bytes_part_received(self, bytes):
60
"""Called when a 'bytes' part is received.
62
A 'bytes' message part can contain any number of bytes. It should not
63
be confused with a 'byte' part, which is always a single byte.
65
raise NotImplementedError(self.bytes_received)
67
def structure_part_received(self, structure):
68
"""Called when a 'structure' part is received.
70
:param structure: some structured data, which will be some combination
71
of list, dict, int, and str objects.
73
raise NotImplementedError(self.bytes_received)
75
def protocol_error(self, exception):
76
"""Called when there is a protocol decoding error.
78
The default implementation just re-raises the exception.
82
def end_received(self):
83
"""Called when the end of the message is received."""
88
class ConventionalRequestHandler(MessageHandler):
89
"""A message handler for "conventional" requests.
91
"Conventional" is used in the sense described in
92
doc/developers/network-protocol.txt: a simple message with arguments and an
96
* args: expecting args
97
* body: expecting body (terminated by receiving a post-body status)
98
* error: expecting post-body error
99
* end: expecting end of message
103
def __init__(self, request_handler, responder):
104
MessageHandler.__init__(self)
105
self.request_handler = request_handler
106
self.responder = responder
107
self.expecting = 'args'
108
self._should_finish_body = False
109
self._response_sent = False
111
def protocol_error(self, exception):
112
if self.responder.response_sent:
113
# We can only send one response to a request, no matter how many
114
# errors happen while processing it.
116
self.responder.send_error(exception)
118
def byte_part_received(self, byte):
119
if not isinstance(byte, bytes):
120
raise TypeError(byte)
121
if self.expecting == 'body':
123
# Success. Nothing more to come except the end of message.
124
self.expecting = 'end'
126
# Error. Expect an error structure.
127
self.expecting = 'error'
129
raise errors.SmartProtocolError(
130
'Non-success status byte in request body: %r' % (byte,))
132
raise errors.SmartProtocolError(
133
'Unexpected message part: byte(%r)' % (byte,))
135
def structure_part_received(self, structure):
136
if self.expecting == 'args':
137
self._args_received(structure)
138
elif self.expecting == 'error':
139
self._error_received(structure)
141
raise errors.SmartProtocolError(
142
'Unexpected message part: structure(%r)' % (structure,))
144
def _args_received(self, args):
145
self.expecting = 'body'
146
self.request_handler.args_received(args)
147
if self.request_handler.finished_reading:
148
self._response_sent = True
149
self.responder.send_response(self.request_handler.response)
150
self.expecting = 'end'
152
def _error_received(self, error_args):
153
self.expecting = 'end'
154
self.request_handler.post_body_error_received(error_args)
156
def bytes_part_received(self, bytes):
157
if self.expecting == 'body':
158
self._should_finish_body = True
159
self.request_handler.accept_body(bytes)
161
raise errors.SmartProtocolError(
162
'Unexpected message part: bytes(%r)' % (bytes,))
164
def end_received(self):
165
if self.expecting not in ['body', 'end']:
166
raise errors.SmartProtocolError(
167
'End of message received prematurely (while expecting %s)'
169
self.expecting = 'nothing'
170
self.request_handler.end_received()
171
if not self.request_handler.finished_reading:
172
raise errors.SmartProtocolError(
173
"Complete conventional request was received, but request "
174
"handler has not finished reading.")
175
if not self._response_sent:
176
self.responder.send_response(self.request_handler.response)
179
class ResponseHandler(object):
180
"""Abstract base class for an object that handles a smart response."""
182
def read_response_tuple(self, expect_body=False):
183
"""Reads and returns the response tuple for the current request.
185
:keyword expect_body: a boolean indicating if a body is expected in the
186
response. Some protocol versions needs this information to know
187
when a response is finished. If False, read_body_bytes should
188
*not* be called afterwards. Defaults to False.
189
:returns: tuple of response arguments.
191
raise NotImplementedError(self.read_response_tuple)
193
def read_body_bytes(self, count=-1):
194
"""Read and return some bytes from the body.
196
:param count: if specified, read up to this many bytes. By default,
197
reads the entire body.
198
:returns: str of bytes from the response body.
200
raise NotImplementedError(self.read_body_bytes)
202
def read_streamed_body(self):
203
"""Returns an iterable that reads and returns a series of body chunks.
205
raise NotImplementedError(self.read_streamed_body)
207
def cancel_read_body(self):
208
"""Stop expecting a body for this response.
210
If expect_body was passed to read_response_tuple, this cancels that
211
expectation (and thus finishes reading the response, allowing a new
212
request to be issued). This is useful if a response turns out to be an
213
error rather than a normal result with a body.
215
raise NotImplementedError(self.cancel_read_body)
218
class ConventionalResponseHandler(MessageHandler, ResponseHandler):
221
MessageHandler.__init__(self)
224
self._bytes_parts = deque()
225
self._body_started = False
226
self._body_stream_status = None
228
self._body_error_args = None
229
self.finished_reading = False
231
def setProtoAndMediumRequest(self, protocol_decoder, medium_request):
232
self._protocol_decoder = protocol_decoder
233
self._medium_request = medium_request
235
def byte_part_received(self, byte):
236
if not isinstance(byte, bytes):
237
raise TypeError(byte)
238
if byte not in [b'E', b'S']:
239
raise errors.SmartProtocolError(
240
'Unknown response status: %r' % (byte,))
241
if self._body_started:
242
if self._body_stream_status is not None:
243
raise errors.SmartProtocolError(
244
'Unexpected byte part received: %r' % (byte,))
245
self._body_stream_status = byte
247
if self.status is not None:
248
raise errors.SmartProtocolError(
249
'Unexpected byte part received: %r' % (byte,))
252
def bytes_part_received(self, bytes):
253
self._body_started = True
254
self._bytes_parts.append(bytes)
256
def structure_part_received(self, structure):
257
if not isinstance(structure, tuple):
258
raise errors.SmartProtocolError(
259
'Args structure is not a sequence: %r' % (structure,))
260
if not self._body_started:
261
if self.args is not None:
262
raise errors.SmartProtocolError(
263
'Unexpected structure received: %r (already got %r)'
264
% (structure, self.args))
265
self.args = structure
267
if self._body_stream_status != b'E':
268
raise errors.SmartProtocolError(
269
'Unexpected structure received after body: %r'
271
self._body_error_args = structure
273
def _wait_for_response_args(self):
274
while self.args is None and not self.finished_reading:
277
def _wait_for_response_end(self):
278
while not self.finished_reading:
281
def _read_more(self):
282
next_read_size = self._protocol_decoder.next_read_size()
283
if next_read_size == 0:
284
# a complete request has been read.
285
self.finished_reading = True
286
self._medium_request.finished_reading()
288
data = self._medium_request.read_bytes(next_read_size)
290
# end of file encountered reading from server
291
if 'hpss' in debug.debug_flags:
293
'decoder state: buf[:10]=%r, state_accept=%s',
294
self._protocol_decoder._get_in_buffer()[:10],
295
self._protocol_decoder.state_accept.__name__)
296
raise errors.ConnectionReset(
297
"Unexpected end of message. "
298
"Please check connectivity and permissions, and report a bug "
299
"if problems persist.")
300
self._protocol_decoder.accept_bytes(data)
302
def protocol_error(self, exception):
303
# Whatever the error is, we're done with this request.
304
self.finished_reading = True
305
self._medium_request.finished_reading()
308
def read_response_tuple(self, expect_body=False):
309
"""Read a response tuple from the wire."""
310
self._wait_for_response_args()
312
self._wait_for_response_end()
313
if 'hpss' in debug.debug_flags:
314
mutter(' result: %r', self.args)
315
if self.status == b'E':
316
self._wait_for_response_end()
317
_raise_smart_server_error(self.args)
318
return tuple(self.args)
320
def read_body_bytes(self, count=-1):
321
"""Read bytes from the body, decoding into a byte stream.
323
We read all bytes at once to ensure we've checked the trailer for
324
errors, and then feed the buffer back as read_body_bytes is called.
326
Like the builtin file.read in Python, a count of -1 (the default) means
327
read the entire body.
329
# TODO: we don't necessarily need to buffer the full request if count
330
# != -1. (2008/04/30, Andrew Bennetts)
331
if self._body is None:
332
self._wait_for_response_end()
333
body_bytes = b''.join(self._bytes_parts)
334
if 'hpss' in debug.debug_flags:
335
mutter(' %d body bytes read', len(body_bytes))
336
self._body = BytesIO(body_bytes)
337
self._bytes_parts = None
338
return self._body.read(count)
340
def read_streamed_body(self):
341
while not self.finished_reading:
342
while self._bytes_parts:
343
bytes_part = self._bytes_parts.popleft()
344
if 'hpssdetail' in debug.debug_flags:
345
mutter(' %d byte part read', len(bytes_part))
348
if self._body_stream_status == b'E':
349
_raise_smart_server_error(self._body_error_args)
351
def cancel_read_body(self):
352
self._wait_for_response_end()
355
def _raise_smart_server_error(error_tuple):
356
"""Raise exception based on tuple received from smart server
358
Specific error translation is handled by breezy.bzr.remote._translate_error
360
if error_tuple[0] == b'UnknownMethod':
361
raise errors.UnknownSmartMethod(error_tuple[1])
362
raise errors.ErrorFromSmartServer(error_tuple)