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 __future__ import absolute_import
19
from bzrlib import lazy_import
20
lazy_import.lazy_import(globals(), """
21
from bzrlib.smart import request as _mod_request
25
from bzrlib.smart import message, protocol
34
class _SmartClient(object):
36
def __init__(self, medium, headers=None):
39
:param medium: a SmartClientMedium
43
self._headers = {'Software version': bzrlib.__version__}
45
self._headers = dict(headers)
48
return '%s(%r)' % (self.__class__.__name__, self._medium)
50
def _call_and_read_response(self, method, args, body=None, readv_body=None,
51
body_stream=None, expect_response_body=True):
52
request = _SmartClientRequest(self, method, args, body=body,
53
readv_body=readv_body, body_stream=body_stream,
54
expect_response_body=expect_response_body)
55
return request.call_and_read_response()
57
def call(self, method, *args):
58
"""Call a method on the remote server."""
59
result, protocol = self.call_expecting_body(method, *args)
60
protocol.cancel_read_body()
63
def call_expecting_body(self, method, *args):
64
"""Call a method and return the result and the protocol object.
66
The body can be read like so::
68
result, smart_protocol = smart_client.call_expecting_body(...)
69
body = smart_protocol.read_body_bytes()
71
return self._call_and_read_response(
72
method, args, expect_response_body=True)
74
def call_with_body_bytes(self, method, args, body):
75
"""Call a method on the remote server with body bytes."""
76
if type(method) is not str:
77
raise TypeError('method must be a byte string, not %r' % (method,))
79
if type(arg) is not str:
80
raise TypeError('args must be byte strings, not %r' % (args,))
81
if type(body) is not str:
82
raise TypeError('body must be byte string, not %r' % (body,))
83
response, response_handler = self._call_and_read_response(
84
method, args, body=body, expect_response_body=False)
87
def call_with_body_bytes_expecting_body(self, method, args, body):
88
"""Call a method on the remote server with body bytes."""
89
if type(method) is not str:
90
raise TypeError('method must be a byte string, not %r' % (method,))
92
if type(arg) is not str:
93
raise TypeError('args must be byte strings, not %r' % (args,))
94
if type(body) is not str:
95
raise TypeError('body must be byte string, not %r' % (body,))
96
response, response_handler = self._call_and_read_response(
97
method, args, body=body, expect_response_body=True)
98
return (response, response_handler)
100
def call_with_body_readv_array(self, args, body):
101
response, response_handler = self._call_and_read_response(
102
args[0], args[1:], readv_body=body, expect_response_body=True)
103
return (response, response_handler)
105
def call_with_body_stream(self, args, stream):
106
response, response_handler = self._call_and_read_response(
107
args[0], args[1:], body_stream=stream,
108
expect_response_body=False)
109
return (response, response_handler)
111
def remote_path_from_transport(self, transport):
112
"""Convert transport into a path suitable for using in a request.
114
Note that the resulting remote path doesn't encode the host name or
115
anything but path, so it is only safe to use it in requests sent over
116
the medium from the matching transport.
118
return self._medium.remote_path_from_transport(transport)
121
class _SmartClientRequest(object):
122
"""Encapsulate the logic for a single request.
124
This class handles things like reconnecting and sending the request a
125
second time when the connection is reset in the middle. It also handles the
126
multiple requests that get made if we don't know what protocol the server
129
Generally, you build up one of these objects, passing in the arguments that
130
you want to send to the server, and then use 'call_and_read_response' to
131
get the response from the server.
134
def __init__(self, client, method, args, body=None, readv_body=None,
135
body_stream=None, expect_response_body=True):
140
self.readv_body = readv_body
141
self.body_stream = body_stream
142
self.expect_response_body = expect_response_body
144
def call_and_read_response(self):
145
"""Send the request to the server, and read the initial response.
147
This doesn't read all of the body content of the response, instead it
148
returns (response_tuple, response_handler). response_tuple is the 'ok',
149
or 'error' information, and 'response_handler' can be used to get the
152
self._run_call_hooks()
153
protocol_version = self.client._medium._protocol_version
154
if protocol_version is None:
155
return self._call_determining_protocol_version()
157
return self._call(protocol_version)
159
def _is_safe_to_send_twice(self):
160
"""Check if the current method is re-entrant safe."""
161
if self.body_stream is not None or 'noretry' in debug.debug_flags:
162
# We can't restart a body stream that has already been consumed.
164
request_type = _mod_request.request_handlers.get_info(self.method)
165
if request_type in ('read', 'idem', 'semi'):
167
# If we have gotten this far, 'stream' cannot be retried, because we
168
# already consumed the local stream.
169
if request_type in ('semivfs', 'mutate', 'stream'):
171
trace.mutter('Unknown request type: %s for method %s'
172
% (request_type, self.method))
175
def _run_call_hooks(self):
176
if not _SmartClient.hooks['call']:
178
params = CallHookParams(self.method, self.args, self.body,
179
self.readv_body, self.client._medium)
180
for hook in _SmartClient.hooks['call']:
183
def _call(self, protocol_version):
184
"""We know the protocol version.
186
So this just sends the request, and then reads the response. This is
187
where the code will be to retry requests if the connection is closed.
189
response_handler = self._send(protocol_version)
191
response_tuple = response_handler.read_response_tuple(
192
expect_body=self.expect_response_body)
193
except errors.ConnectionReset, e:
194
self.client._medium.reset()
195
if not self._is_safe_to_send_twice():
197
trace.warning('ConnectionReset reading response for %r, retrying'
199
trace.log_exception_quietly()
200
encoder, response_handler = self._construct_protocol(
202
self._send_no_retry(encoder)
203
response_tuple = response_handler.read_response_tuple(
204
expect_body=self.expect_response_body)
205
return (response_tuple, response_handler)
207
def _call_determining_protocol_version(self):
208
"""Determine what protocol the remote server supports.
210
We do this by placing a request in the most recent protocol, and
211
handling the UnexpectedProtocolVersionMarker from the server.
213
for protocol_version in [3, 2]:
214
if protocol_version == 2:
215
# If v3 doesn't work, the remote side is older than 1.6.
216
self.client._medium._remember_remote_is_before((1, 6))
218
response_tuple, response_handler = self._call(protocol_version)
219
except errors.UnexpectedProtocolVersionMarker, err:
220
# TODO: We could recover from this without disconnecting if
221
# we recognise the protocol version.
223
'Server does not understand Bazaar network protocol %d,'
224
' reconnecting. (Upgrade the server to avoid this.)'
225
% (protocol_version,))
226
self.client._medium.disconnect()
228
except errors.ErrorFromSmartServer:
229
# If we received an error reply from the server, then it
230
# must be ok with this protocol version.
231
self.client._medium._protocol_version = protocol_version
234
self.client._medium._protocol_version = protocol_version
235
return response_tuple, response_handler
236
raise errors.SmartProtocolError(
237
'Server is not a Bazaar server: ' + str(err))
239
def _construct_protocol(self, version):
240
"""Build the encoding stack for a given protocol version."""
241
request = self.client._medium.get_request()
243
request_encoder = protocol.ProtocolThreeRequester(request)
244
response_handler = message.ConventionalResponseHandler()
245
response_proto = protocol.ProtocolThreeDecoder(
246
response_handler, expect_version_marker=True)
247
response_handler.setProtoAndMediumRequest(response_proto, request)
249
request_encoder = protocol.SmartClientRequestProtocolTwo(request)
250
response_handler = request_encoder
252
request_encoder = protocol.SmartClientRequestProtocolOne(request)
253
response_handler = request_encoder
254
return request_encoder, response_handler
256
def _send(self, protocol_version):
257
"""Encode the request, and send it to the server.
259
This will retry a request if we get a ConnectionReset while sending the
260
request to the server. (Unless we have a body_stream that we have
261
already started consuming, since we can't restart body_streams)
263
:return: response_handler as defined by _construct_protocol
265
encoder, response_handler = self._construct_protocol(protocol_version)
267
self._send_no_retry(encoder)
268
except errors.ConnectionReset, e:
269
# If we fail during the _send_no_retry phase, then we can
270
# be confident that the server did not get our request, because we
271
# haven't started waiting for the reply yet. So try the request
272
# again. We only issue a single retry, because if the connection
273
# really is down, there is no reason to loop endlessly.
275
# Connection is dead, so close our end of it.
276
self.client._medium.reset()
277
if (('noretry' in debug.debug_flags)
278
or (self.body_stream is not None
279
and encoder.body_stream_started)):
280
# We can't restart a body_stream that has been partially
281
# consumed, so we don't retry.
282
# Note: We don't have to worry about
283
# SmartClientRequestProtocolOne or Two, because they don't
284
# support client-side body streams.
286
trace.warning('ConnectionReset calling %r, retrying'
288
trace.log_exception_quietly()
289
encoder, response_handler = self._construct_protocol(
291
self._send_no_retry(encoder)
292
return response_handler
294
def _send_no_retry(self, encoder):
295
"""Just encode the request and try to send it."""
296
encoder.set_headers(self.client._headers)
297
if self.body is not None:
298
if self.readv_body is not None:
299
raise AssertionError(
300
"body and readv_body are mutually exclusive.")
301
if self.body_stream is not None:
302
raise AssertionError(
303
"body and body_stream are mutually exclusive.")
304
encoder.call_with_body_bytes((self.method, ) + self.args, self.body)
305
elif self.readv_body is not None:
306
if self.body_stream is not None:
307
raise AssertionError(
308
"readv_body and body_stream are mutually exclusive.")
309
encoder.call_with_body_readv_array((self.method, ) + self.args,
311
elif self.body_stream is not None:
312
encoder.call_with_body_stream((self.method, ) + self.args,
315
encoder.call(self.method, *self.args)
318
class SmartClientHooks(hooks.Hooks):
321
hooks.Hooks.__init__(self, "bzrlib.smart.client", "_SmartClient.hooks")
322
self.add_hook('call',
323
"Called when the smart client is submitting a request to the "
324
"smart server. Called with a bzrlib.smart.client.CallHookParams "
325
"object. Streaming request bodies, and responses, are not "
329
_SmartClient.hooks = SmartClientHooks()
332
class CallHookParams(object):
334
def __init__(self, method, args, body, readv_body, medium):
338
self.readv_body = readv_body
342
attrs = dict((k, v) for (k, v) in self.__dict__.iteritems()
344
return '<%s %r>' % (self.__class__.__name__, attrs)
346
def __eq__(self, other):
347
if type(other) is not type(self):
348
return NotImplemented
349
return self.__dict__ == other.__dict__
351
def __ne__(self, other):
352
return not self == other
added
removed
bzrlib/smart/client.py
