1
# Copyright (C) 2006-2010 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 bzrlib import lazy_import
18
lazy_import.lazy_import(globals(), """
19
from bzrlib.smart import request as _mod_request
23
from bzrlib.smart import message, protocol
32
class _SmartClient(object):
34
def __init__(self, medium, headers=None):
37
:param medium: a SmartClientMedium
41
self._headers = {'Software version': bzrlib.__version__}
43
self._headers = dict(headers)
46
return '%s(%r)' % (self.__class__.__name__, self._medium)
48
def _call_and_read_response(self, method, args, body=None, readv_body=None,
49
body_stream=None, expect_response_body=True):
50
request = _SmartClientRequest(self, method, args, body=body,
51
readv_body=readv_body, body_stream=body_stream,
52
expect_response_body=expect_response_body)
53
return request.call_and_read_response()
55
def call(self, method, *args):
56
"""Call a method on the remote server."""
57
result, protocol = self.call_expecting_body(method, *args)
58
protocol.cancel_read_body()
61
def call_expecting_body(self, method, *args):
62
"""Call a method and return the result and the protocol object.
64
The body can be read like so::
66
result, smart_protocol = smart_client.call_expecting_body(...)
67
body = smart_protocol.read_body_bytes()
69
return self._call_and_read_response(
70
method, args, expect_response_body=True)
72
def call_with_body_bytes(self, method, args, body):
73
"""Call a method on the remote server with body bytes."""
74
if type(method) is not str:
75
raise TypeError('method must be a byte string, not %r' % (method,))
77
if type(arg) is not str:
78
raise TypeError('args must be byte strings, not %r' % (args,))
79
if type(body) is not str:
80
raise TypeError('body must be byte string, not %r' % (body,))
81
response, response_handler = self._call_and_read_response(
82
method, args, body=body, expect_response_body=False)
85
def call_with_body_bytes_expecting_body(self, method, args, body):
86
"""Call a method on the remote server with body bytes."""
87
if type(method) is not str:
88
raise TypeError('method must be a byte string, not %r' % (method,))
90
if type(arg) is not str:
91
raise TypeError('args must be byte strings, not %r' % (args,))
92
if type(body) is not str:
93
raise TypeError('body must be byte string, not %r' % (body,))
94
response, response_handler = self._call_and_read_response(
95
method, args, body=body, expect_response_body=True)
96
return (response, response_handler)
98
def call_with_body_readv_array(self, args, body):
99
response, response_handler = self._call_and_read_response(
100
args[0], args[1:], readv_body=body, expect_response_body=True)
101
return (response, response_handler)
103
def call_with_body_stream(self, args, stream):
104
response, response_handler = self._call_and_read_response(
105
args[0], args[1:], body_stream=stream,
106
expect_response_body=False)
107
return (response, response_handler)
109
def remote_path_from_transport(self, transport):
110
"""Convert transport into a path suitable for using in a request.
112
Note that the resulting remote path doesn't encode the host name or
113
anything but path, so it is only safe to use it in requests sent over
114
the medium from the matching transport.
116
return self._medium.remote_path_from_transport(transport)
119
class _SmartClientRequest(object):
120
"""Encapsulate the logic for a single request.
122
This class handles things like reconnecting and sending the request a
123
second time when the connection is reset in the middle. It also handles the
124
multiple requests that get made if we don't know what protocol the server
127
Generally, you build up one of these objects, passing in the arguments that
128
you want to send to the server, and then use 'call_and_read_response' to
129
get the response from the server.
132
def __init__(self, client, method, args, body=None, readv_body=None,
133
body_stream=None, expect_response_body=True):
138
self.readv_body = readv_body
139
self.body_stream = body_stream
140
self.expect_response_body = expect_response_body
142
def call_and_read_response(self):
143
"""Send the request to the server, and read the initial response.
145
This doesn't read all of the body content of the response, instead it
146
returns (response_tuple, response_handler). response_tuple is the 'ok',
147
or 'error' information, and 'response_handler' can be used to get the
150
self._run_call_hooks()
151
protocol_version = self.client._medium._protocol_version
152
if protocol_version is None:
153
return self._call_determining_protocol_version()
155
return self._call(protocol_version)
157
def _is_safe_to_send_twice(self):
158
"""Check if the current method is re-entrant safe."""
159
if self.body_stream is not None or 'noretry' in debug.debug_flags:
160
# We can't restart a body stream that has already been consumed.
162
request_type = _mod_request.request_handlers.get_info(self.method)
163
if request_type in ('read', 'idem', 'semi'):
165
# If we have gotten this far, 'stream' cannot be retried, because we
166
# already consumed the local stream.
167
if request_type in ('semivfs', 'mutate', 'stream'):
169
trace.mutter('Unknown request type: %s for method %s'
170
% (request_type, self.method))
173
def _run_call_hooks(self):
174
if not _SmartClient.hooks['call']:
176
params = CallHookParams(self.method, self.args, self.body,
177
self.readv_body, self.client._medium)
178
for hook in _SmartClient.hooks['call']:
181
def _call(self, protocol_version):
182
"""We know the protocol version.
184
So this just sends the request, and then reads the response. This is
185
where the code will be to retry requests if the connection is closed.
187
response_handler = self._send(protocol_version)
189
response_tuple = response_handler.read_response_tuple(
190
expect_body=self.expect_response_body)
191
except errors.ConnectionReset, e:
192
self.client._medium.reset()
193
if not self._is_safe_to_send_twice():
195
trace.warning('ConnectionReset reading response for %r, retrying'
197
trace.log_exception_quietly()
198
encoder, response_handler = self._construct_protocol(
200
self._send_no_retry(encoder)
201
response_tuple = response_handler.read_response_tuple(
202
expect_body=self.expect_response_body)
203
return (response_tuple, response_handler)
205
def _call_determining_protocol_version(self):
206
"""Determine what protocol the remote server supports.
208
We do this by placing a request in the most recent protocol, and
209
handling the UnexpectedProtocolVersionMarker from the server.
211
for protocol_version in [3, 2]:
212
if protocol_version == 2:
213
# If v3 doesn't work, the remote side is older than 1.6.
214
self.client._medium._remember_remote_is_before((1, 6))
216
response_tuple, response_handler = self._call(protocol_version)
217
except errors.UnexpectedProtocolVersionMarker, err:
218
# TODO: We could recover from this without disconnecting if
219
# we recognise the protocol version.
221
'Server does not understand Bazaar network protocol %d,'
222
' reconnecting. (Upgrade the server to avoid this.)'
223
% (protocol_version,))
224
self.client._medium.disconnect()
226
except errors.ErrorFromSmartServer:
227
# If we received an error reply from the server, then it
228
# must be ok with this protocol version.
229
self.client._medium._protocol_version = protocol_version
232
self.client._medium._protocol_version = protocol_version
233
return response_tuple, response_handler
234
raise errors.SmartProtocolError(
235
'Server is not a Bazaar server: ' + str(err))
237
def _construct_protocol(self, version):
238
"""Build the encoding stack for a given protocol version."""
239
request = self.client._medium.get_request()
241
request_encoder = protocol.ProtocolThreeRequester(request)
242
response_handler = message.ConventionalResponseHandler()
243
response_proto = protocol.ProtocolThreeDecoder(
244
response_handler, expect_version_marker=True)
245
response_handler.setProtoAndMediumRequest(response_proto, request)
247
request_encoder = protocol.SmartClientRequestProtocolTwo(request)
248
response_handler = request_encoder
250
request_encoder = protocol.SmartClientRequestProtocolOne(request)
251
response_handler = request_encoder
252
return request_encoder, response_handler
254
def _send(self, protocol_version):
255
"""Encode the request, and send it to the server.
257
This will retry a request if we get a ConnectionReset while sending the
258
request to the server. (Unless we have a body_stream that we have
259
already started consuming, since we can't restart body_streams)
261
:return: response_handler as defined by _construct_protocol
263
encoder, response_handler = self._construct_protocol(protocol_version)
265
self._send_no_retry(encoder)
266
except errors.ConnectionReset, e:
267
# If we fail during the _send_no_retry phase, then we can
268
# be confident that the server did not get our request, because we
269
# haven't started waiting for the reply yet. So try the request
270
# again. We only issue a single retry, because if the connection
271
# really is down, there is no reason to loop endlessly.
273
# Connection is dead, so close our end of it.
274
self.client._medium.reset()
275
if (('noretry' in debug.debug_flags)
276
or (self.body_stream is not None
277
and encoder.body_stream_started)):
278
# We can't restart a body_stream that has been partially
279
# consumed, so we don't retry.
280
# Note: We don't have to worry about
281
# SmartClientRequestProtocolOne or Two, because they don't
282
# support client-side body streams.
284
trace.warning('ConnectionReset calling %r, retrying'
286
trace.log_exception_quietly()
287
encoder, response_handler = self._construct_protocol(
289
self._send_no_retry(encoder)
290
return response_handler
292
def _send_no_retry(self, encoder):
293
"""Just encode the request and try to send it."""
294
encoder.set_headers(self.client._headers)
295
if self.body is not None:
296
if self.readv_body is not None:
297
raise AssertionError(
298
"body and readv_body are mutually exclusive.")
299
if self.body_stream is not None:
300
raise AssertionError(
301
"body and body_stream are mutually exclusive.")
302
encoder.call_with_body_bytes((self.method, ) + self.args, self.body)
303
elif self.readv_body is not None:
304
if self.body_stream is not None:
305
raise AssertionError(
306
"readv_body and body_stream are mutually exclusive.")
307
encoder.call_with_body_readv_array((self.method, ) + self.args,
309
elif self.body_stream is not None:
310
encoder.call_with_body_stream((self.method, ) + self.args,
313
encoder.call(self.method, *self.args)
316
class SmartClientHooks(hooks.Hooks):
319
hooks.Hooks.__init__(self, "bzrlib.smart.client", "_SmartClient.hooks")
320
self.add_hook('call',
321
"Called when the smart client is submitting a request to the "
322
"smart server. Called with a bzrlib.smart.client.CallHookParams "
323
"object. Streaming request bodies, and responses, are not "
327
_SmartClient.hooks = SmartClientHooks()
330
class CallHookParams(object):
332
def __init__(self, method, args, body, readv_body, medium):
336
self.readv_body = readv_body
340
attrs = dict((k, v) for (k, v) in self.__dict__.iteritems()
342
return '<%s %r>' % (self.__class__.__name__, attrs)
344
def __eq__(self, other):
345
if type(other) is not type(self):
346
return NotImplemented
347
return self.__dict__ == other.__dict__
349
def __ne__(self, other):
350
return not self == other
added
removed
bzrlib/smart/client.py
