/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/bzr/smart/client.py

  • Committer: Jelmer Vernooij
  • Date: 2020-02-18 01:57:45 UTC
  • mto: This revision was merged to the branch mainline in revision 7493.
  • Revision ID: jelmer@jelmer.uk-20200218015745-q2ss9tsk74h4nh61
drop use of future.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/bzr/smart/client.py
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
 
import bzrlib
18
 
from bzrlib.smart import message, protocol
19
 
from bzrlib.trace import warning
20
 
from bzrlib import (
 
17
from ... import lazy_import
 
18
lazy_import.lazy_import(globals(), """
 
19
from breezy.bzr.smart import request as _mod_request
 
20
""")
 
21
 
 
22
import breezy
 
23
from . import message, protocol
 
24
from ... import (
 
25
    debug,
21
26
    errors,
22
27
    hooks,
 
28
    trace,
23
29
    )
24
30
 
25
31
 
32
38
        """
33
39
        self._medium = medium
34
40
        if headers is None:
35
 
            self._headers = {'Software version': bzrlib.__version__}
 
41
            self._headers = {
 
42
                b'Software version': breezy.__version__.encode('utf-8')}
36
43
        else:
37
44
            self._headers = dict(headers)
38
45
 
39
46
    def __repr__(self):
40
47
        return '%s(%r)' % (self.__class__.__name__, self._medium)
41
48
 
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(
45
 
            protocol_version)
46
 
        encoder.set_headers(self._headers)
47
 
        if body is not None:
48
 
            if readv_body is not None:
49
 
                raise AssertionError(
50
 
                    "body and readv_body are mutually exclusive.")
51
 
            if body_stream is not None:
52
 
                raise AssertionError(
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:
57
 
                raise AssertionError(
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)
62
 
        else:
63
 
            encoder.call(method, *args)
64
 
        return response_handler
65
 
 
66
 
    def _run_call_hooks(self, method, args, body, readv_body):
67
 
        if not _SmartClient.hooks['call']:
68
 
            return
69
 
        params = CallHookParams(method, args, body, readv_body, self._medium)
70
 
        for hook in _SmartClient.hooks['call']:
71
 
            hook(params)
72
 
 
73
49
    def _call_and_read_response(self, method, args, body=None, readv_body=None,
74
 
            body_stream=None, expect_response_body=True):
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),
82
 
                    response_handler)
83
 
        else:
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)
91
 
                try:
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.
97
 
                    warning(
98
 
                        'Server does not understand Bazaar network protocol %d,'
99
 
                        ' reconnecting.  (Upgrade the server to avoid this.)'
100
 
                        % (protocol_version,))
101
 
                    self._medium.disconnect()
102
 
                    continue
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
107
 
                    raise
108
 
                else:
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))
113
 
 
114
 
    def _construct_protocol(self, version):
115
 
        request = self._medium.get_request()
116
 
        if version == 3:
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)
122
 
        elif version == 2:
123
 
            request_encoder = protocol.SmartClientRequestProtocolTwo(request)
124
 
            response_handler = request_encoder
125
 
        else:
126
 
            request_encoder = protocol.SmartClientRequestProtocolOne(request)
127
 
            response_handler = request_encoder
128
 
        return request_encoder, response_handler
 
50
                                body_stream=None, expect_response_body=True):
 
51
        request = _SmartClientRequest(self, method, args, body=body,
 
52
                                      readv_body=readv_body, body_stream=body_stream,
 
53
                                      expect_response_body=expect_response_body)
 
54
        return request.call_and_read_response()
129
55
 
130
56
    def call(self, method, *args):
131
57
        """Call a method on the remote server."""
146
72
 
147
73
    def call_with_body_bytes(self, method, args, body):
148
74
        """Call a method on the remote server with body bytes."""
149
 
        if type(method) is not str:
 
75
        if not isinstance(method, bytes):
150
76
            raise TypeError('method must be a byte string, not %r' % (method,))
151
77
        for arg in args:
152
 
            if type(arg) is not str:
 
78
            if not isinstance(arg, bytes):
153
79
                raise TypeError('args must be byte strings, not %r' % (args,))
154
 
        if type(body) is not str:
 
80
        if not isinstance(body, bytes):
155
81
            raise TypeError('body must be byte string, not %r' % (body,))
156
82
        response, response_handler = self._call_and_read_response(
157
83
            method, args, body=body, expect_response_body=False)
159
85
 
160
86
    def call_with_body_bytes_expecting_body(self, method, args, body):
161
87
        """Call a method on the remote server with body bytes."""
162
 
        if type(method) is not str:
 
88
        if not isinstance(method, bytes):
163
89
            raise TypeError('method must be a byte string, not %r' % (method,))
164
90
        for arg in args:
165
 
            if type(arg) is not str:
 
91
            if not isinstance(arg, bytes):
166
92
                raise TypeError('args must be byte strings, not %r' % (args,))
167
 
        if type(body) is not str:
 
93
        if not isinstance(body, bytes):
168
94
            raise TypeError('body must be byte string, not %r' % (body,))
169
95
        response, response_handler = self._call_and_read_response(
170
96
            method, args, body=body, expect_response_body=True)
172
98
 
173
99
    def call_with_body_readv_array(self, args, body):
174
100
        response, response_handler = self._call_and_read_response(
175
 
                args[0], args[1:], readv_body=body, expect_response_body=True)
 
101
            args[0], args[1:], readv_body=body, expect_response_body=True)
176
102
        return (response, response_handler)
177
103
 
178
104
    def call_with_body_stream(self, args, stream):
179
105
        response, response_handler = self._call_and_read_response(
180
 
                args[0], args[1:], body_stream=stream,
181
 
                expect_response_body=False)
 
106
            args[0], args[1:], body_stream=stream,
 
107
            expect_response_body=False)
182
108
        return (response, response_handler)
183
109
 
184
110
    def remote_path_from_transport(self, transport):
188
114
        anything but path, so it is only safe to use it in requests sent over
189
115
        the medium from the matching transport.
190
116
        """
191
 
        return self._medium.remote_path_from_transport(transport)
 
117
        return self._medium.remote_path_from_transport(transport).encode('utf-8')
 
118
 
 
119
 
 
120
class _SmartClientRequest(object):
 
121
    """Encapsulate the logic for a single request.
 
122
 
 
123
    This class handles things like reconnecting and sending the request a
 
124
    second time when the connection is reset in the middle. It also handles the
 
125
    multiple requests that get made if we don't know what protocol the server
 
126
    supports yet.
 
127
 
 
128
    Generally, you build up one of these objects, passing in the arguments that
 
129
    you want to send to the server, and then use 'call_and_read_response' to
 
130
    get the response from the server.
 
131
    """
 
132
 
 
133
    def __init__(self, client, method, args, body=None, readv_body=None,
 
134
                 body_stream=None, expect_response_body=True):
 
135
        self.client = client
 
136
        self.method = method
 
137
        self.args = args
 
138
        self.body = body
 
139
        self.readv_body = readv_body
 
140
        self.body_stream = body_stream
 
141
        self.expect_response_body = expect_response_body
 
142
 
 
143
    def call_and_read_response(self):
 
144
        """Send the request to the server, and read the initial response.
 
145
 
 
146
        This doesn't read all of the body content of the response, instead it
 
147
        returns (response_tuple, response_handler). response_tuple is the 'ok',
 
148
        or 'error' information, and 'response_handler' can be used to get the
 
149
        content stream out.
 
150
        """
 
151
        self._run_call_hooks()
 
152
        protocol_version = self.client._medium._protocol_version
 
153
        if protocol_version is None:
 
154
            return self._call_determining_protocol_version()
 
155
        else:
 
156
            return self._call(protocol_version)
 
157
 
 
158
    def _is_safe_to_send_twice(self):
 
159
        """Check if the current method is re-entrant safe."""
 
160
        if self.body_stream is not None or 'noretry' in debug.debug_flags:
 
161
            # We can't restart a body stream that has already been consumed.
 
162
            return False
 
163
        request_type = _mod_request.request_handlers.get_info(self.method)
 
164
        if request_type in ('read', 'idem', 'semi'):
 
165
            return True
 
166
        # If we have gotten this far, 'stream' cannot be retried, because we
 
167
        # already consumed the local stream.
 
168
        if request_type in ('semivfs', 'mutate', 'stream'):
 
169
            return False
 
170
        trace.mutter('Unknown request type: %s for method %s'
 
171
                     % (request_type, self.method))
 
172
        return False
 
173
 
 
174
    def _run_call_hooks(self):
 
175
        if not _SmartClient.hooks['call']:
 
176
            return
 
177
        params = CallHookParams(self.method, self.args, self.body,
 
178
                                self.readv_body, self.client._medium)
 
179
        for hook in _SmartClient.hooks['call']:
 
180
            hook(params)
 
181
 
 
182
    def _call(self, protocol_version):
 
183
        """We know the protocol version.
 
184
 
 
185
        So this just sends the request, and then reads the response. This is
 
186
        where the code will be to retry requests if the connection is closed.
 
187
        """
 
188
        response_handler = self._send(protocol_version)
 
189
        try:
 
190
            response_tuple = response_handler.read_response_tuple(
 
191
                expect_body=self.expect_response_body)
 
192
        except errors.ConnectionReset as e:
 
193
            self.client._medium.reset()
 
194
            if not self._is_safe_to_send_twice():
 
195
                raise
 
196
            trace.warning('ConnectionReset reading response for %r, retrying'
 
197
                          % (self.method,))
 
198
            trace.log_exception_quietly()
 
199
            encoder, response_handler = self._construct_protocol(
 
200
                protocol_version)
 
201
            self._send_no_retry(encoder)
 
202
            response_tuple = response_handler.read_response_tuple(
 
203
                expect_body=self.expect_response_body)
 
204
        return (response_tuple, response_handler)
 
205
 
 
206
    def _call_determining_protocol_version(self):
 
207
        """Determine what protocol the remote server supports.
 
208
 
 
209
        We do this by placing a request in the most recent protocol, and
 
210
        handling the UnexpectedProtocolVersionMarker from the server.
 
211
        """
 
212
        last_err = None
 
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))
 
217
            try:
 
218
                response_tuple, response_handler = self._call(protocol_version)
 
219
            except errors.UnexpectedProtocolVersionMarker as err:
 
220
                # TODO: We could recover from this without disconnecting if
 
221
                # we recognise the protocol version.
 
222
                trace.warning(
 
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()
 
227
                last_err = err
 
228
                continue
 
229
            except errors.ErrorFromSmartServer:
 
230
                # If we received an error reply from the server, then it
 
231
                # must be ok with this protocol version.
 
232
                self.client._medium._protocol_version = protocol_version
 
233
                raise
 
234
            else:
 
235
                self.client._medium._protocol_version = protocol_version
 
236
                return response_tuple, response_handler
 
237
        raise errors.SmartProtocolError(
 
238
            'Server is not a Bazaar server: ' + str(last_err))
 
239
 
 
240
    def _construct_protocol(self, version):
 
241
        """Build the encoding stack for a given protocol version."""
 
242
        request = self.client._medium.get_request()
 
243
        if version == 3:
 
244
            request_encoder = protocol.ProtocolThreeRequester(request)
 
245
            response_handler = message.ConventionalResponseHandler()
 
246
            response_proto = protocol.ProtocolThreeDecoder(
 
247
                response_handler, expect_version_marker=True)
 
248
            response_handler.setProtoAndMediumRequest(response_proto, request)
 
249
        elif version == 2:
 
250
            request_encoder = protocol.SmartClientRequestProtocolTwo(request)
 
251
            response_handler = request_encoder
 
252
        else:
 
253
            request_encoder = protocol.SmartClientRequestProtocolOne(request)
 
254
            response_handler = request_encoder
 
255
        return request_encoder, response_handler
 
256
 
 
257
    def _send(self, protocol_version):
 
258
        """Encode the request, and send it to the server.
 
259
 
 
260
        This will retry a request if we get a ConnectionReset while sending the
 
261
        request to the server. (Unless we have a body_stream that we have
 
262
        already started consuming, since we can't restart body_streams)
 
263
 
 
264
        :return: response_handler as defined by _construct_protocol
 
265
        """
 
266
        encoder, response_handler = self._construct_protocol(protocol_version)
 
267
        try:
 
268
            self._send_no_retry(encoder)
 
269
        except errors.ConnectionReset as e:
 
270
            # If we fail during the _send_no_retry phase, then we can
 
271
            # be confident that the server did not get our request, because we
 
272
            # haven't started waiting for the reply yet. So try the request
 
273
            # again. We only issue a single retry, because if the connection
 
274
            # really is down, there is no reason to loop endlessly.
 
275
 
 
276
            # Connection is dead, so close our end of it.
 
277
            self.client._medium.reset()
 
278
            if (('noretry' in debug.debug_flags) or
 
279
                (self.body_stream is not None and
 
280
                    encoder.body_stream_started)):
 
281
                # We can't restart a body_stream that has been partially
 
282
                # consumed, so we don't retry.
 
283
                # Note: We don't have to worry about
 
284
                #   SmartClientRequestProtocolOne or Two, because they don't
 
285
                #   support client-side body streams.
 
286
                raise
 
287
            trace.warning('ConnectionReset calling %r, retrying'
 
288
                          % (self.method,))
 
289
            trace.log_exception_quietly()
 
290
            encoder, response_handler = self._construct_protocol(
 
291
                protocol_version)
 
292
            self._send_no_retry(encoder)
 
293
        return response_handler
 
294
 
 
295
    def _send_no_retry(self, encoder):
 
296
        """Just encode the request and try to send it."""
 
297
        encoder.set_headers(self.client._headers)
 
298
        if self.body is not None:
 
299
            if self.readv_body is not None:
 
300
                raise AssertionError(
 
301
                    "body and readv_body are mutually exclusive.")
 
302
            if self.body_stream is not None:
 
303
                raise AssertionError(
 
304
                    "body and body_stream are mutually exclusive.")
 
305
            encoder.call_with_body_bytes(
 
306
                (self.method, ) + self.args, self.body)
 
307
        elif self.readv_body is not None:
 
308
            if self.body_stream is not None:
 
309
                raise AssertionError(
 
310
                    "readv_body and body_stream are mutually exclusive.")
 
311
            encoder.call_with_body_readv_array((self.method, ) + self.args,
 
312
                                               self.readv_body)
 
313
        elif self.body_stream is not None:
 
314
            encoder.call_with_body_stream((self.method, ) + self.args,
 
315
                                          self.body_stream)
 
316
        else:
 
317
            encoder.call(self.method, *self.args)
192
318
 
193
319
 
194
320
class SmartClientHooks(hooks.Hooks):
195
321
 
196
322
    def __init__(self):
197
 
        hooks.Hooks.__init__(self)
198
 
        self.create_hook(hooks.HookPoint('call',
199
 
            "Called when the smart client is submitting a request to the "
200
 
            "smart server. Called with a bzrlib.smart.client.CallHookParams "
201
 
            "object. Streaming request bodies, and responses, are not "
202
 
            "accessible.", None, None))
 
323
        hooks.Hooks.__init__(
 
324
            self, "breezy.bzr.smart.client", "_SmartClient.hooks")
 
325
        self.add_hook('call',
 
326
                      "Called when the smart client is submitting a request to the "
 
327
                      "smart server. Called with a breezy.bzr.smart.client.CallHookParams "
 
328
                      "object. Streaming request bodies, and responses, are not "
 
329
                      "accessible.", None)
203
330
 
204
331
 
205
332
_SmartClient.hooks = SmartClientHooks()
215
342
        self.medium = medium
216
343
 
217
344
    def __repr__(self):
218
 
        attrs = dict((k, v) for (k, v) in self.__dict__.iteritems()
 
345
        attrs = dict((k, v) for k, v in self.__dict__.items()
219
346
                     if v is not None)
220
347
        return '<%s %r>' % (self.__class__.__name__, attrs)
221
348
 
222
349
    def __eq__(self, other):
223
 
        if type(other) is not type(self):
 
350
        if not isinstance(other, type(self)):
224
351
            return NotImplemented
225
352
        return self.__dict__ == other.__dict__
226
353