47
39
def __repr__(self):
48
40
return '%s(%r)' % (self.__class__.__name__, self._medium)
42
def _send_request(self, protocol_version, method, args, body=None,
43
readv_body=None, body_stream=None):
44
encoder, response_handler = self._construct_protocol(
46
encoder.set_headers(self._headers)
48
if readv_body is not None:
50
"body and readv_body are mutually exclusive.")
51
if body_stream is not None:
53
"body and body_stream are mutually exclusive.")
54
encoder.call_with_body_bytes((method, ) + args, body)
55
elif readv_body is not None:
56
if body_stream is not None:
58
"readv_body and body_stream are mutually exclusive.")
59
encoder.call_with_body_readv_array((method, ) + args, readv_body)
60
elif body_stream is not None:
61
encoder.call_with_body_stream((method, ) + args, body_stream)
63
encoder.call(method, *args)
64
return response_handler
66
def _run_call_hooks(self, method, args, body, readv_body):
67
if not _SmartClient.hooks['call']:
69
params = CallHookParams(method, args, body, readv_body, self._medium)
70
for hook in _SmartClient.hooks['call']:
50
73
def _call_and_read_response(self, method, args, body=None, readv_body=None,
51
74
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()
75
self._run_call_hooks(method, args, body, readv_body)
76
if self._medium._protocol_version is not None:
77
response_handler = self._send_request(
78
self._medium._protocol_version, method, args, body=body,
79
readv_body=readv_body, body_stream=body_stream)
80
return (response_handler.read_response_tuple(
81
expect_body=expect_response_body),
84
for protocol_version in [3, 2]:
85
if protocol_version == 2:
86
# If v3 doesn't work, the remote side is older than 1.6.
87
self._medium._remember_remote_is_before((1, 6))
88
response_handler = self._send_request(
89
protocol_version, method, args, body=body,
90
readv_body=readv_body, body_stream=body_stream)
92
response_tuple = response_handler.read_response_tuple(
93
expect_body=expect_response_body)
94
except errors.UnexpectedProtocolVersionMarker, err:
95
# TODO: We could recover from this without disconnecting if
96
# we recognise the protocol version.
98
'Server does not understand Bazaar network protocol %d,'
99
' reconnecting. (Upgrade the server to avoid this.)'
100
% (protocol_version,))
101
self._medium.disconnect()
103
except errors.ErrorFromSmartServer:
104
# If we received an error reply from the server, then it
105
# must be ok with this protocol version.
106
self._medium._protocol_version = protocol_version
109
self._medium._protocol_version = protocol_version
110
return response_tuple, response_handler
111
raise errors.SmartProtocolError(
112
'Server is not a Bazaar server: ' + str(err))
114
def _construct_protocol(self, version):
115
request = self._medium.get_request()
117
request_encoder = protocol.ProtocolThreeRequester(request)
118
response_handler = message.ConventionalResponseHandler()
119
response_proto = protocol.ProtocolThreeDecoder(
120
response_handler, expect_version_marker=True)
121
response_handler.setProtoAndMediumRequest(response_proto, request)
123
request_encoder = protocol.SmartClientRequestProtocolTwo(request)
124
response_handler = request_encoder
126
request_encoder = protocol.SmartClientRequestProtocolOne(request)
127
response_handler = request_encoder
128
return request_encoder, response_handler
57
130
def call(self, method, *args):
58
131
"""Call a method on the remote server."""
118
191
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
194
class SmartClientHooks(hooks.Hooks):
320
196
def __init__(self):
321
hooks.Hooks.__init__(self, "bzrlib.smart.client", "_SmartClient.hooks")
322
self.add_hook('call',
197
hooks.Hooks.__init__(self)
198
self.create_hook(hooks.HookPoint('call',
323
199
"Called when the smart client is submitting a request to the "
324
200
"smart server. Called with a bzrlib.smart.client.CallHookParams "
325
201
"object. Streaming request bodies, and responses, are not "
202
"accessible.", None, None))
329
205
_SmartClient.hooks = SmartClientHooks()
added
removed
bzrlib/smart/client.py
