~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: Robert Collins
  • Date: 2005-10-03 01:15:02 UTC
  • mfrom: (1092.2.28)
  • Revision ID: robertc@robertcollins.net-20051003011502-f579a509a136b774
mergeĀ fromĀ baz2bzr

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2006 Canonical Ltd
2
 
#
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.
7
 
#
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.
12
 
#
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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
 
 
17
 
"""The 'medium' layer for the smart servers and clients.
18
 
 
19
 
"Medium" here is the noun meaning "a means of transmission", not the adjective
20
 
for "the quality between big and small."
21
 
 
22
 
Media carry the bytes of the requests somehow (e.g. via TCP, wrapped in HTTP, or
23
 
over SSH), and pass them to and from the protocol logic.  See the overview in
24
 
bzrlib/transport/smart/__init__.py.
25
 
"""
26
 
 
27
 
import errno
28
 
import os
29
 
import socket
30
 
import sys
31
 
import urllib
32
 
 
33
 
from bzrlib.lazy_import import lazy_import
34
 
lazy_import(globals(), """
35
 
import atexit
36
 
import weakref
37
 
from bzrlib import (
38
 
    debug,
39
 
    errors,
40
 
    osutils,
41
 
    symbol_versioning,
42
 
    trace,
43
 
    urlutils,
44
 
    )
45
 
from bzrlib.smart import client, protocol
46
 
from bzrlib.transport import ssh
47
 
""")
48
 
 
49
 
 
50
 
# We must not read any more than 64k at a time so we don't risk "no buffer
51
 
# space available" errors on some platforms.  Windows in particular is likely
52
 
# to give error 10053 or 10055 if we read more than 64k from a socket.
53
 
_MAX_READ_SIZE = 64 * 1024
54
 
 
55
 
 
56
 
def _get_protocol_factory_for_bytes(bytes):
57
 
    """Determine the right protocol factory for 'bytes'.
58
 
 
59
 
    This will return an appropriate protocol factory depending on the version
60
 
    of the protocol being used, as determined by inspecting the given bytes.
61
 
    The bytes should have at least one newline byte (i.e. be a whole line),
62
 
    otherwise it's possible that a request will be incorrectly identified as
63
 
    version 1.
64
 
 
65
 
    Typical use would be::
66
 
 
67
 
         factory, unused_bytes = _get_protocol_factory_for_bytes(bytes)
68
 
         server_protocol = factory(transport, write_func, root_client_path)
69
 
         server_protocol.accept_bytes(unused_bytes)
70
 
 
71
 
    :param bytes: a str of bytes of the start of the request.
72
 
    :returns: 2-tuple of (protocol_factory, unused_bytes).  protocol_factory is
73
 
        a callable that takes three args: transport, write_func,
74
 
        root_client_path.  unused_bytes are any bytes that were not part of a
75
 
        protocol version marker.
76
 
    """
77
 
    if bytes.startswith(protocol.MESSAGE_VERSION_THREE):
78
 
        protocol_factory = protocol.build_server_protocol_three
79
 
        bytes = bytes[len(protocol.MESSAGE_VERSION_THREE):]
80
 
    elif bytes.startswith(protocol.REQUEST_VERSION_TWO):
81
 
        protocol_factory = protocol.SmartServerRequestProtocolTwo
82
 
        bytes = bytes[len(protocol.REQUEST_VERSION_TWO):]
83
 
    else:
84
 
        protocol_factory = protocol.SmartServerRequestProtocolOne
85
 
    return protocol_factory, bytes
86
 
 
87
 
 
88
 
def _get_line(read_bytes_func):
89
 
    """Read bytes using read_bytes_func until a newline byte.
90
 
    
91
 
    This isn't particularly efficient, so should only be used when the
92
 
    expected size of the line is quite short.
93
 
    
94
 
    :returns: a tuple of two strs: (line, excess)
95
 
    """
96
 
    newline_pos = -1
97
 
    bytes = ''
98
 
    while newline_pos == -1:
99
 
        new_bytes = read_bytes_func(1)
100
 
        bytes += new_bytes
101
 
        if new_bytes == '':
102
 
            # Ran out of bytes before receiving a complete line.
103
 
            return bytes, ''
104
 
        newline_pos = bytes.find('\n')
105
 
    line = bytes[:newline_pos+1]
106
 
    excess = bytes[newline_pos+1:]
107
 
    return line, excess
108
 
 
109
 
 
110
 
class SmartMedium(object):
111
 
    """Base class for smart protocol media, both client- and server-side."""
112
 
 
113
 
    def __init__(self):
114
 
        self._push_back_buffer = None
115
 
        
116
 
    def _push_back(self, bytes):
117
 
        """Return unused bytes to the medium, because they belong to the next
118
 
        request(s).
119
 
 
120
 
        This sets the _push_back_buffer to the given bytes.
121
 
        """
122
 
        if self._push_back_buffer is not None:
123
 
            raise AssertionError(
124
 
                "_push_back called when self._push_back_buffer is %r"
125
 
                % (self._push_back_buffer,))
126
 
        if bytes == '':
127
 
            return
128
 
        self._push_back_buffer = bytes
129
 
 
130
 
    def _get_push_back_buffer(self):
131
 
        if self._push_back_buffer == '':
132
 
            raise AssertionError(
133
 
                '%s._push_back_buffer should never be the empty string, '
134
 
                'which can be confused with EOF' % (self,))
135
 
        bytes = self._push_back_buffer
136
 
        self._push_back_buffer = None
137
 
        return bytes
138
 
 
139
 
    def read_bytes(self, desired_count):
140
 
        """Read some bytes from this medium.
141
 
 
142
 
        :returns: some bytes, possibly more or less than the number requested
143
 
            in 'desired_count' depending on the medium.
144
 
        """
145
 
        if self._push_back_buffer is not None:
146
 
            return self._get_push_back_buffer()
147
 
        bytes_to_read = min(desired_count, _MAX_READ_SIZE)
148
 
        return self._read_bytes(bytes_to_read)
149
 
 
150
 
    def _read_bytes(self, count):
151
 
        raise NotImplementedError(self._read_bytes)
152
 
 
153
 
    def _get_line(self):
154
 
        """Read bytes from this request's response until a newline byte.
155
 
        
156
 
        This isn't particularly efficient, so should only be used when the
157
 
        expected size of the line is quite short.
158
 
 
159
 
        :returns: a string of bytes ending in a newline (byte 0x0A).
160
 
        """
161
 
        line, excess = _get_line(self.read_bytes)
162
 
        self._push_back(excess)
163
 
        return line
164
 
 
165
 
 
166
 
class SmartServerStreamMedium(SmartMedium):
167
 
    """Handles smart commands coming over a stream.
168
 
 
169
 
    The stream may be a pipe connected to sshd, or a tcp socket, or an
170
 
    in-process fifo for testing.
171
 
 
172
 
    One instance is created for each connected client; it can serve multiple
173
 
    requests in the lifetime of the connection.
174
 
 
175
 
    The server passes requests through to an underlying backing transport, 
176
 
    which will typically be a LocalTransport looking at the server's filesystem.
177
 
 
178
 
    :ivar _push_back_buffer: a str of bytes that have been read from the stream
179
 
        but not used yet, or None if there are no buffered bytes.  Subclasses
180
 
        should make sure to exhaust this buffer before reading more bytes from
181
 
        the stream.  See also the _push_back method.
182
 
    """
183
 
 
184
 
    def __init__(self, backing_transport, root_client_path='/'):
185
 
        """Construct new server.
186
 
 
187
 
        :param backing_transport: Transport for the directory served.
188
 
        """
189
 
        # backing_transport could be passed to serve instead of __init__
190
 
        self.backing_transport = backing_transport
191
 
        self.root_client_path = root_client_path
192
 
        self.finished = False
193
 
        SmartMedium.__init__(self)
194
 
 
195
 
    def serve(self):
196
 
        """Serve requests until the client disconnects."""
197
 
        # Keep a reference to stderr because the sys module's globals get set to
198
 
        # None during interpreter shutdown.
199
 
        from sys import stderr
200
 
        try:
201
 
            while not self.finished:
202
 
                server_protocol = self._build_protocol()
203
 
                self._serve_one_request(server_protocol)
204
 
        except Exception, e:
205
 
            stderr.write("%s terminating on exception %s\n" % (self, e))
206
 
            raise
207
 
 
208
 
    def _build_protocol(self):
209
 
        """Identifies the version of the incoming request, and returns an
210
 
        a protocol object that can interpret it.
211
 
 
212
 
        If more bytes than the version prefix of the request are read, they will
213
 
        be fed into the protocol before it is returned.
214
 
 
215
 
        :returns: a SmartServerRequestProtocol.
216
 
        """
217
 
        bytes = self._get_line()
218
 
        protocol_factory, unused_bytes = _get_protocol_factory_for_bytes(bytes)
219
 
        protocol = protocol_factory(
220
 
            self.backing_transport, self._write_out, self.root_client_path)
221
 
        protocol.accept_bytes(unused_bytes)
222
 
        return protocol
223
 
 
224
 
    def _serve_one_request(self, protocol):
225
 
        """Read one request from input, process, send back a response.
226
 
        
227
 
        :param protocol: a SmartServerRequestProtocol.
228
 
        """
229
 
        try:
230
 
            self._serve_one_request_unguarded(protocol)
231
 
        except KeyboardInterrupt:
232
 
            raise
233
 
        except Exception, e:
234
 
            self.terminate_due_to_error()
235
 
 
236
 
    def terminate_due_to_error(self):
237
 
        """Called when an unhandled exception from the protocol occurs."""
238
 
        raise NotImplementedError(self.terminate_due_to_error)
239
 
 
240
 
    def _read_bytes(self, desired_count):
241
 
        """Get some bytes from the medium.
242
 
 
243
 
        :param desired_count: number of bytes we want to read.
244
 
        """
245
 
        raise NotImplementedError(self._read_bytes)
246
 
 
247
 
 
248
 
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
249
 
 
250
 
    def __init__(self, sock, backing_transport, root_client_path='/'):
251
 
        """Constructor.
252
 
 
253
 
        :param sock: the socket the server will read from.  It will be put
254
 
            into blocking mode.
255
 
        """
256
 
        SmartServerStreamMedium.__init__(
257
 
            self, backing_transport, root_client_path=root_client_path)
258
 
        sock.setblocking(True)
259
 
        self.socket = sock
260
 
 
261
 
    def _serve_one_request_unguarded(self, protocol):
262
 
        while protocol.next_read_size():
263
 
            # We can safely try to read large chunks.  If there is less data
264
 
            # than _MAX_READ_SIZE ready, the socket wil just return a short
265
 
            # read immediately rather than block.
266
 
            bytes = self.read_bytes(_MAX_READ_SIZE)
267
 
            if bytes == '':
268
 
                self.finished = True
269
 
                return
270
 
            protocol.accept_bytes(bytes)
271
 
        
272
 
        self._push_back(protocol.unused_data)
273
 
 
274
 
    def _read_bytes(self, desired_count):
275
 
        # We ignore the desired_count because on sockets it's more efficient to
276
 
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
277
 
        return osutils.until_no_eintr(self.socket.recv, _MAX_READ_SIZE)
278
 
 
279
 
    def terminate_due_to_error(self):
280
 
        # TODO: This should log to a server log file, but no such thing
281
 
        # exists yet.  Andrew Bennetts 2006-09-29.
282
 
        self.socket.close()
283
 
        self.finished = True
284
 
 
285
 
    def _write_out(self, bytes):
286
 
        osutils.send_all(self.socket, bytes)
287
 
 
288
 
 
289
 
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
290
 
 
291
 
    def __init__(self, in_file, out_file, backing_transport):
292
 
        """Construct new server.
293
 
 
294
 
        :param in_file: Python file from which requests can be read.
295
 
        :param out_file: Python file to write responses.
296
 
        :param backing_transport: Transport for the directory served.
297
 
        """
298
 
        SmartServerStreamMedium.__init__(self, backing_transport)
299
 
        if sys.platform == 'win32':
300
 
            # force binary mode for files
301
 
            import msvcrt
302
 
            for f in (in_file, out_file):
303
 
                fileno = getattr(f, 'fileno', None)
304
 
                if fileno:
305
 
                    msvcrt.setmode(fileno(), os.O_BINARY)
306
 
        self._in = in_file
307
 
        self._out = out_file
308
 
 
309
 
    def _serve_one_request_unguarded(self, protocol):
310
 
        while True:
311
 
            # We need to be careful not to read past the end of the current
312
 
            # request, or else the read from the pipe will block, so we use
313
 
            # protocol.next_read_size().
314
 
            bytes_to_read = protocol.next_read_size()
315
 
            if bytes_to_read == 0:
316
 
                # Finished serving this request.
317
 
                self._out.flush()
318
 
                return
319
 
            bytes = self.read_bytes(bytes_to_read)
320
 
            if bytes == '':
321
 
                # Connection has been closed.
322
 
                self.finished = True
323
 
                self._out.flush()
324
 
                return
325
 
            protocol.accept_bytes(bytes)
326
 
 
327
 
    def _read_bytes(self, desired_count):
328
 
        return self._in.read(desired_count)
329
 
 
330
 
    def terminate_due_to_error(self):
331
 
        # TODO: This should log to a server log file, but no such thing
332
 
        # exists yet.  Andrew Bennetts 2006-09-29.
333
 
        self._out.close()
334
 
        self.finished = True
335
 
 
336
 
    def _write_out(self, bytes):
337
 
        self._out.write(bytes)
338
 
 
339
 
 
340
 
class SmartClientMediumRequest(object):
341
 
    """A request on a SmartClientMedium.
342
 
 
343
 
    Each request allows bytes to be provided to it via accept_bytes, and then
344
 
    the response bytes to be read via read_bytes.
345
 
 
346
 
    For instance:
347
 
    request.accept_bytes('123')
348
 
    request.finished_writing()
349
 
    result = request.read_bytes(3)
350
 
    request.finished_reading()
351
 
 
352
 
    It is up to the individual SmartClientMedium whether multiple concurrent
353
 
    requests can exist. See SmartClientMedium.get_request to obtain instances 
354
 
    of SmartClientMediumRequest, and the concrete Medium you are using for 
355
 
    details on concurrency and pipelining.
356
 
    """
357
 
 
358
 
    def __init__(self, medium):
359
 
        """Construct a SmartClientMediumRequest for the medium medium."""
360
 
        self._medium = medium
361
 
        # we track state by constants - we may want to use the same
362
 
        # pattern as BodyReader if it gets more complex.
363
 
        # valid states are: "writing", "reading", "done"
364
 
        self._state = "writing"
365
 
 
366
 
    def accept_bytes(self, bytes):
367
 
        """Accept bytes for inclusion in this request.
368
 
 
369
 
        This method may not be be called after finished_writing() has been
370
 
        called.  It depends upon the Medium whether or not the bytes will be
371
 
        immediately transmitted. Message based Mediums will tend to buffer the
372
 
        bytes until finished_writing() is called.
373
 
 
374
 
        :param bytes: A bytestring.
375
 
        """
376
 
        if self._state != "writing":
377
 
            raise errors.WritingCompleted(self)
378
 
        self._accept_bytes(bytes)
379
 
 
380
 
    def _accept_bytes(self, bytes):
381
 
        """Helper for accept_bytes.
382
 
 
383
 
        Accept_bytes checks the state of the request to determing if bytes
384
 
        should be accepted. After that it hands off to _accept_bytes to do the
385
 
        actual acceptance.
386
 
        """
387
 
        raise NotImplementedError(self._accept_bytes)
388
 
 
389
 
    def finished_reading(self):
390
 
        """Inform the request that all desired data has been read.
391
 
 
392
 
        This will remove the request from the pipeline for its medium (if the
393
 
        medium supports pipelining) and any further calls to methods on the
394
 
        request will raise ReadingCompleted.
395
 
        """
396
 
        if self._state == "writing":
397
 
            raise errors.WritingNotComplete(self)
398
 
        if self._state != "reading":
399
 
            raise errors.ReadingCompleted(self)
400
 
        self._state = "done"
401
 
        self._finished_reading()
402
 
 
403
 
    def _finished_reading(self):
404
 
        """Helper for finished_reading.
405
 
 
406
 
        finished_reading checks the state of the request to determine if 
407
 
        finished_reading is allowed, and if it is hands off to _finished_reading
408
 
        to perform the action.
409
 
        """
410
 
        raise NotImplementedError(self._finished_reading)
411
 
 
412
 
    def finished_writing(self):
413
 
        """Finish the writing phase of this request.
414
 
 
415
 
        This will flush all pending data for this request along the medium.
416
 
        After calling finished_writing, you may not call accept_bytes anymore.
417
 
        """
418
 
        if self._state != "writing":
419
 
            raise errors.WritingCompleted(self)
420
 
        self._state = "reading"
421
 
        self._finished_writing()
422
 
 
423
 
    def _finished_writing(self):
424
 
        """Helper for finished_writing.
425
 
 
426
 
        finished_writing checks the state of the request to determine if 
427
 
        finished_writing is allowed, and if it is hands off to _finished_writing
428
 
        to perform the action.
429
 
        """
430
 
        raise NotImplementedError(self._finished_writing)
431
 
 
432
 
    def read_bytes(self, count):
433
 
        """Read bytes from this requests response.
434
 
 
435
 
        This method will block and wait for count bytes to be read. It may not
436
 
        be invoked until finished_writing() has been called - this is to ensure
437
 
        a message-based approach to requests, for compatibility with message
438
 
        based mediums like HTTP.
439
 
        """
440
 
        if self._state == "writing":
441
 
            raise errors.WritingNotComplete(self)
442
 
        if self._state != "reading":
443
 
            raise errors.ReadingCompleted(self)
444
 
        return self._read_bytes(count)
445
 
 
446
 
    def _read_bytes(self, count):
447
 
        """Helper for SmartClientMediumRequest.read_bytes.
448
 
 
449
 
        read_bytes checks the state of the request to determing if bytes
450
 
        should be read. After that it hands off to _read_bytes to do the
451
 
        actual read.
452
 
        
453
 
        By default this forwards to self._medium.read_bytes because we are
454
 
        operating on the medium's stream.
455
 
        """
456
 
        return self._medium.read_bytes(count)
457
 
 
458
 
    def read_line(self):
459
 
        line = self._read_line()
460
 
        if not line.endswith('\n'):
461
 
            # end of file encountered reading from server
462
 
            raise errors.ConnectionReset(
463
 
                "please check connectivity and permissions",
464
 
                "(and try -Dhpss if further diagnosis is required)")
465
 
        return line
466
 
 
467
 
    def _read_line(self):
468
 
        """Helper for SmartClientMediumRequest.read_line.
469
 
        
470
 
        By default this forwards to self._medium._get_line because we are
471
 
        operating on the medium's stream.
472
 
        """
473
 
        return self._medium._get_line()
474
 
 
475
 
 
476
 
class _DebugCounter(object):
477
 
    """An object that counts the HPSS calls made to each client medium.
478
 
 
479
 
    When a medium is garbage-collected, or failing that when atexit functions
480
 
    are run, the total number of calls made on that medium are reported via
481
 
    trace.note.
482
 
    """
483
 
 
484
 
    def __init__(self):
485
 
        self.counts = weakref.WeakKeyDictionary()
486
 
        client._SmartClient.hooks.install_named_hook(
487
 
            'call', self.increment_call_count, 'hpss call counter')
488
 
        atexit.register(self.flush_all)
489
 
 
490
 
    def track(self, medium):
491
 
        """Start tracking calls made to a medium.
492
 
 
493
 
        This only keeps a weakref to the medium, so shouldn't affect the
494
 
        medium's lifetime.
495
 
        """
496
 
        medium_repr = repr(medium)
497
 
        # Add this medium to the WeakKeyDictionary
498
 
        self.counts[medium] = [0, medium_repr]
499
 
        # Weakref callbacks are fired in reverse order of their association
500
 
        # with the referenced object.  So we add a weakref *after* adding to
501
 
        # the WeakKeyDict so that we can report the value from it before the
502
 
        # entry is removed by the WeakKeyDict's own callback.
503
 
        ref = weakref.ref(medium, self.done)
504
 
 
505
 
    def increment_call_count(self, params):
506
 
        # Increment the count in the WeakKeyDictionary
507
 
        value = self.counts[params.medium]
508
 
        value[0] += 1
509
 
 
510
 
    def done(self, ref):
511
 
        value = self.counts[ref]
512
 
        count, medium_repr = value
513
 
        # In case this callback is invoked for the same ref twice (by the
514
 
        # weakref callback and by the atexit function), set the call count back
515
 
        # to 0 so this item won't be reported twice.
516
 
        value[0] = 0
517
 
        if count != 0:
518
 
            trace.note('HPSS calls: %d %s', count, medium_repr)
519
 
        
520
 
    def flush_all(self):
521
 
        for ref in list(self.counts.keys()):
522
 
            self.done(ref)
523
 
 
524
 
_debug_counter = None
525
 
  
526
 
  
527
 
class SmartClientMedium(SmartMedium):
528
 
    """Smart client is a medium for sending smart protocol requests over."""
529
 
 
530
 
    def __init__(self, base):
531
 
        super(SmartClientMedium, self).__init__()
532
 
        self.base = base
533
 
        self._protocol_version_error = None
534
 
        self._protocol_version = None
535
 
        self._done_hello = False
536
 
        # Be optimistic: we assume the remote end can accept new remote
537
 
        # requests until we get an error saying otherwise.
538
 
        # _remote_version_is_before tracks the bzr version the remote side
539
 
        # can be based on what we've seen so far.
540
 
        self._remote_version_is_before = None
541
 
        # Install debug hook function if debug flag is set.
542
 
        if 'hpss' in debug.debug_flags:
543
 
            global _debug_counter
544
 
            if _debug_counter is None:
545
 
                _debug_counter = _DebugCounter()
546
 
            _debug_counter.track(self)
547
 
 
548
 
    def _is_remote_before(self, version_tuple):
549
 
        """Is it possible the remote side supports RPCs for a given version?
550
 
 
551
 
        Typical use::
552
 
 
553
 
            needed_version = (1, 2)
554
 
            if medium._is_remote_before(needed_version):
555
 
                fallback_to_pre_1_2_rpc()
556
 
            else:
557
 
                try:
558
 
                    do_1_2_rpc()
559
 
                except UnknownSmartMethod:
560
 
                    medium._remember_remote_is_before(needed_version)
561
 
                    fallback_to_pre_1_2_rpc()
562
 
 
563
 
        :seealso: _remember_remote_is_before
564
 
        """
565
 
        if self._remote_version_is_before is None:
566
 
            # So far, the remote side seems to support everything
567
 
            return False
568
 
        return version_tuple >= self._remote_version_is_before
569
 
 
570
 
    def _remember_remote_is_before(self, version_tuple):
571
 
        """Tell this medium that the remote side is older the given version.
572
 
 
573
 
        :seealso: _is_remote_before
574
 
        """
575
 
        if (self._remote_version_is_before is not None and
576
 
            version_tuple > self._remote_version_is_before):
577
 
            raise AssertionError(
578
 
                "_remember_remote_is_before(%r) called, but "
579
 
                "_remember_remote_is_before(%r) was called previously."
580
 
                % (version_tuple, self._remote_version_is_before))
581
 
        self._remote_version_is_before = version_tuple
582
 
 
583
 
    def protocol_version(self):
584
 
        """Find out if 'hello' smart request works."""
585
 
        if self._protocol_version_error is not None:
586
 
            raise self._protocol_version_error
587
 
        if not self._done_hello:
588
 
            try:
589
 
                medium_request = self.get_request()
590
 
                # Send a 'hello' request in protocol version one, for maximum
591
 
                # backwards compatibility.
592
 
                client_protocol = protocol.SmartClientRequestProtocolOne(medium_request)
593
 
                client_protocol.query_version()
594
 
                self._done_hello = True
595
 
            except errors.SmartProtocolError, e:
596
 
                # Cache the error, just like we would cache a successful
597
 
                # result.
598
 
                self._protocol_version_error = e
599
 
                raise
600
 
        return '2'
601
 
 
602
 
    def should_probe(self):
603
 
        """Should RemoteBzrDirFormat.probe_transport send a smart request on
604
 
        this medium?
605
 
 
606
 
        Some transports are unambiguously smart-only; there's no need to check
607
 
        if the transport is able to carry smart requests, because that's all
608
 
        it is for.  In those cases, this method should return False.
609
 
 
610
 
        But some HTTP transports can sometimes fail to carry smart requests,
611
 
        but still be usuable for accessing remote bzrdirs via plain file
612
 
        accesses.  So for those transports, their media should return True here
613
 
        so that RemoteBzrDirFormat can determine if it is appropriate for that
614
 
        transport.
615
 
        """
616
 
        return False
617
 
 
618
 
    def disconnect(self):
619
 
        """If this medium maintains a persistent connection, close it.
620
 
        
621
 
        The default implementation does nothing.
622
 
        """
623
 
        
624
 
    def remote_path_from_transport(self, transport):
625
 
        """Convert transport into a path suitable for using in a request.
626
 
        
627
 
        Note that the resulting remote path doesn't encode the host name or
628
 
        anything but path, so it is only safe to use it in requests sent over
629
 
        the medium from the matching transport.
630
 
        """
631
 
        medium_base = urlutils.join(self.base, '/')
632
 
        rel_url = urlutils.relative_url(medium_base, transport.base)
633
 
        return urllib.unquote(rel_url)
634
 
 
635
 
 
636
 
class SmartClientStreamMedium(SmartClientMedium):
637
 
    """Stream based medium common class.
638
 
 
639
 
    SmartClientStreamMediums operate on a stream. All subclasses use a common
640
 
    SmartClientStreamMediumRequest for their requests, and should implement
641
 
    _accept_bytes and _read_bytes to allow the request objects to send and
642
 
    receive bytes.
643
 
    """
644
 
 
645
 
    def __init__(self, base):
646
 
        SmartClientMedium.__init__(self, base)
647
 
        self._current_request = None
648
 
 
649
 
    def accept_bytes(self, bytes):
650
 
        self._accept_bytes(bytes)
651
 
 
652
 
    def __del__(self):
653
 
        """The SmartClientStreamMedium knows how to close the stream when it is
654
 
        finished with it.
655
 
        """
656
 
        self.disconnect()
657
 
 
658
 
    def _flush(self):
659
 
        """Flush the output stream.
660
 
        
661
 
        This method is used by the SmartClientStreamMediumRequest to ensure that
662
 
        all data for a request is sent, to avoid long timeouts or deadlocks.
663
 
        """
664
 
        raise NotImplementedError(self._flush)
665
 
 
666
 
    def get_request(self):
667
 
        """See SmartClientMedium.get_request().
668
 
 
669
 
        SmartClientStreamMedium always returns a SmartClientStreamMediumRequest
670
 
        for get_request.
671
 
        """
672
 
        return SmartClientStreamMediumRequest(self)
673
 
 
674
 
 
675
 
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
676
 
    """A client medium using simple pipes.
677
 
    
678
 
    This client does not manage the pipes: it assumes they will always be open.
679
 
    """
680
 
 
681
 
    def __init__(self, readable_pipe, writeable_pipe, base):
682
 
        SmartClientStreamMedium.__init__(self, base)
683
 
        self._readable_pipe = readable_pipe
684
 
        self._writeable_pipe = writeable_pipe
685
 
 
686
 
    def _accept_bytes(self, bytes):
687
 
        """See SmartClientStreamMedium.accept_bytes."""
688
 
        self._writeable_pipe.write(bytes)
689
 
 
690
 
    def _flush(self):
691
 
        """See SmartClientStreamMedium._flush()."""
692
 
        self._writeable_pipe.flush()
693
 
 
694
 
    def _read_bytes(self, count):
695
 
        """See SmartClientStreamMedium._read_bytes."""
696
 
        return self._readable_pipe.read(count)
697
 
 
698
 
 
699
 
class SmartSSHClientMedium(SmartClientStreamMedium):
700
 
    """A client medium using SSH."""
701
 
    
702
 
    def __init__(self, host, port=None, username=None, password=None,
703
 
            base=None, vendor=None, bzr_remote_path=None):
704
 
        """Creates a client that will connect on the first use.
705
 
        
706
 
        :param vendor: An optional override for the ssh vendor to use. See
707
 
            bzrlib.transport.ssh for details on ssh vendors.
708
 
        """
709
 
        SmartClientStreamMedium.__init__(self, base)
710
 
        self._connected = False
711
 
        self._host = host
712
 
        self._password = password
713
 
        self._port = port
714
 
        self._username = username
715
 
        self._read_from = None
716
 
        self._ssh_connection = None
717
 
        self._vendor = vendor
718
 
        self._write_to = None
719
 
        self._bzr_remote_path = bzr_remote_path
720
 
        if self._bzr_remote_path is None:
721
 
            symbol_versioning.warn(
722
 
                'bzr_remote_path is required as of bzr 0.92',
723
 
                DeprecationWarning, stacklevel=2)
724
 
            self._bzr_remote_path = os.environ.get('BZR_REMOTE_PATH', 'bzr')
725
 
 
726
 
    def _accept_bytes(self, bytes):
727
 
        """See SmartClientStreamMedium.accept_bytes."""
728
 
        self._ensure_connection()
729
 
        self._write_to.write(bytes)
730
 
 
731
 
    def disconnect(self):
732
 
        """See SmartClientMedium.disconnect()."""
733
 
        if not self._connected:
734
 
            return
735
 
        self._read_from.close()
736
 
        self._write_to.close()
737
 
        self._ssh_connection.close()
738
 
        self._connected = False
739
 
 
740
 
    def _ensure_connection(self):
741
 
        """Connect this medium if not already connected."""
742
 
        if self._connected:
743
 
            return
744
 
        if self._vendor is None:
745
 
            vendor = ssh._get_ssh_vendor()
746
 
        else:
747
 
            vendor = self._vendor
748
 
        self._ssh_connection = vendor.connect_ssh(self._username,
749
 
                self._password, self._host, self._port,
750
 
                command=[self._bzr_remote_path, 'serve', '--inet',
751
 
                         '--directory=/', '--allow-writes'])
752
 
        self._read_from, self._write_to = \
753
 
            self._ssh_connection.get_filelike_channels()
754
 
        self._connected = True
755
 
 
756
 
    def _flush(self):
757
 
        """See SmartClientStreamMedium._flush()."""
758
 
        self._write_to.flush()
759
 
 
760
 
    def _read_bytes(self, count):
761
 
        """See SmartClientStreamMedium.read_bytes."""
762
 
        if not self._connected:
763
 
            raise errors.MediumNotConnected(self)
764
 
        bytes_to_read = min(count, _MAX_READ_SIZE)
765
 
        return self._read_from.read(bytes_to_read)
766
 
 
767
 
 
768
 
# Port 4155 is the default port for bzr://, registered with IANA.
769
 
BZR_DEFAULT_INTERFACE = None
770
 
BZR_DEFAULT_PORT = 4155
771
 
 
772
 
 
773
 
class SmartTCPClientMedium(SmartClientStreamMedium):
774
 
    """A client medium using TCP."""
775
 
    
776
 
    def __init__(self, host, port, base):
777
 
        """Creates a client that will connect on the first use."""
778
 
        SmartClientStreamMedium.__init__(self, base)
779
 
        self._connected = False
780
 
        self._host = host
781
 
        self._port = port
782
 
        self._socket = None
783
 
 
784
 
    def _accept_bytes(self, bytes):
785
 
        """See SmartClientMedium.accept_bytes."""
786
 
        self._ensure_connection()
787
 
        osutils.send_all(self._socket, bytes)
788
 
 
789
 
    def disconnect(self):
790
 
        """See SmartClientMedium.disconnect()."""
791
 
        if not self._connected:
792
 
            return
793
 
        self._socket.close()
794
 
        self._socket = None
795
 
        self._connected = False
796
 
 
797
 
    def _ensure_connection(self):
798
 
        """Connect this medium if not already connected."""
799
 
        if self._connected:
800
 
            return
801
 
        if self._port is None:
802
 
            port = BZR_DEFAULT_PORT
803
 
        else:
804
 
            port = int(self._port)
805
 
        try:
806
 
            sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC, 
807
 
                socket.SOCK_STREAM, 0, 0)
808
 
        except socket.gaierror, (err_num, err_msg):
809
 
            raise errors.ConnectionError("failed to lookup %s:%d: %s" %
810
 
                    (self._host, port, err_msg))
811
 
        # Initialize err in case there are no addresses returned:
812
 
        err = socket.error("no address found for %s" % self._host)
813
 
        for (family, socktype, proto, canonname, sockaddr) in sockaddrs:
814
 
            try:
815
 
                self._socket = socket.socket(family, socktype, proto)
816
 
                self._socket.setsockopt(socket.IPPROTO_TCP, 
817
 
                                        socket.TCP_NODELAY, 1)
818
 
                self._socket.connect(sockaddr)
819
 
            except socket.error, err:
820
 
                if self._socket is not None:
821
 
                    self._socket.close()
822
 
                self._socket = None
823
 
                continue
824
 
            break
825
 
        if self._socket is None:
826
 
            # socket errors either have a (string) or (errno, string) as their
827
 
            # args.
828
 
            if type(err.args) is str:
829
 
                err_msg = err.args
830
 
            else:
831
 
                err_msg = err.args[1]
832
 
            raise errors.ConnectionError("failed to connect to %s:%d: %s" %
833
 
                    (self._host, port, err_msg))
834
 
        self._connected = True
835
 
 
836
 
    def _flush(self):
837
 
        """See SmartClientStreamMedium._flush().
838
 
        
839
 
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and 
840
 
        add a means to do a flush, but that can be done in the future.
841
 
        """
842
 
 
843
 
    def _read_bytes(self, count):
844
 
        """See SmartClientMedium.read_bytes."""
845
 
        if not self._connected:
846
 
            raise errors.MediumNotConnected(self)
847
 
        # We ignore the desired_count because on sockets it's more efficient to
848
 
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
849
 
        try:
850
 
            return self._socket.recv(_MAX_READ_SIZE)
851
 
        except socket.error, e:
852
 
            if len(e.args) and e.args[0] == errno.ECONNRESET:
853
 
                # Callers expect an empty string in that case
854
 
                return ''
855
 
            else:
856
 
                raise
857
 
 
858
 
 
859
 
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
860
 
    """A SmartClientMediumRequest that works with an SmartClientStreamMedium."""
861
 
 
862
 
    def __init__(self, medium):
863
 
        SmartClientMediumRequest.__init__(self, medium)
864
 
        # check that we are safe concurrency wise. If some streams start
865
 
        # allowing concurrent requests - i.e. via multiplexing - then this
866
 
        # assert should be moved to SmartClientStreamMedium.get_request,
867
 
        # and the setting/unsetting of _current_request likewise moved into
868
 
        # that class : but its unneeded overhead for now. RBC 20060922
869
 
        if self._medium._current_request is not None:
870
 
            raise errors.TooManyConcurrentRequests(self._medium)
871
 
        self._medium._current_request = self
872
 
 
873
 
    def _accept_bytes(self, bytes):
874
 
        """See SmartClientMediumRequest._accept_bytes.
875
 
        
876
 
        This forwards to self._medium._accept_bytes because we are operating
877
 
        on the mediums stream.
878
 
        """
879
 
        self._medium._accept_bytes(bytes)
880
 
 
881
 
    def _finished_reading(self):
882
 
        """See SmartClientMediumRequest._finished_reading.
883
 
 
884
 
        This clears the _current_request on self._medium to allow a new 
885
 
        request to be created.
886
 
        """
887
 
        if self._medium._current_request is not self:
888
 
            raise AssertionError()
889
 
        self._medium._current_request = None
890
 
        
891
 
    def _finished_writing(self):
892
 
        """See SmartClientMediumRequest._finished_writing.
893
 
 
894
 
        This invokes self._medium._flush to ensure all bytes are transmitted.
895
 
        """
896
 
        self._medium._flush()
897