22
22
Media carry the bytes of the requests somehow (e.g. via TCP, wrapped in HTTP, or
23
23
over SSH), and pass them to and from the protocol logic. See the overview in
24
breezy/transport/smart/__init__.py.
24
bzrlib/transport/smart/__init__.py.
27
from __future__ import absolute_import
36
from ...lazy_import import lazy_import
37
lazy_import(globals(), """
50
from breezy.i18n import gettext
51
from breezy.bzr.smart import client, protocol, request, signals, vfs
52
from breezy.transport import ssh
59
# Throughout this module buffer size parameters are either limited to be at
60
# most _MAX_READ_SIZE, or are ignored and _MAX_READ_SIZE is used instead.
61
# For this module's purposes, MAX_SOCKET_CHUNK is a reasonable size for reads
62
# from non-sockets as well.
63
_MAX_READ_SIZE = osutils.MAX_SOCKET_CHUNK
66
class HpssVfsRequestNotAllowed(errors.BzrError):
68
_fmt = ("VFS requests over the smart server are not allowed. Encountered: "
69
"%(method)s, %(arguments)s.")
71
def __init__(self, method, arguments):
73
self.arguments = arguments
76
def _get_protocol_factory_for_bytes(bytes):
77
"""Determine the right protocol factory for 'bytes'.
79
This will return an appropriate protocol factory depending on the version
80
of the protocol being used, as determined by inspecting the given bytes.
81
The bytes should have at least one newline byte (i.e. be a whole line),
82
otherwise it's possible that a request will be incorrectly identified as
85
Typical use would be::
87
factory, unused_bytes = _get_protocol_factory_for_bytes(bytes)
88
server_protocol = factory(transport, write_func, root_client_path)
89
server_protocol.accept_bytes(unused_bytes)
91
:param bytes: a str of bytes of the start of the request.
92
:returns: 2-tuple of (protocol_factory, unused_bytes). protocol_factory is
93
a callable that takes three args: transport, write_func,
94
root_client_path. unused_bytes are any bytes that were not part of a
95
protocol version marker.
97
if bytes.startswith(protocol.MESSAGE_VERSION_THREE):
98
protocol_factory = protocol.build_server_protocol_three
99
bytes = bytes[len(protocol.MESSAGE_VERSION_THREE):]
100
elif bytes.startswith(protocol.REQUEST_VERSION_TWO):
101
protocol_factory = protocol.SmartServerRequestProtocolTwo
102
bytes = bytes[len(protocol.REQUEST_VERSION_TWO):]
104
protocol_factory = protocol.SmartServerRequestProtocolOne
105
return protocol_factory, bytes
108
def _get_line(read_bytes_func):
109
"""Read bytes using read_bytes_func until a newline byte.
111
This isn't particularly efficient, so should only be used when the
112
expected size of the line is quite short.
114
:returns: a tuple of two strs: (line, excess)
118
while newline_pos == -1:
119
new_bytes = read_bytes_func(1)
122
# Ran out of bytes before receiving a complete line.
124
newline_pos = bytes.find(b'\n')
125
line = bytes[:newline_pos+1]
126
excess = bytes[newline_pos+1:]
130
class SmartMedium(object):
131
"""Base class for smart protocol media, both client- and server-side."""
134
self._push_back_buffer = None
136
def _push_back(self, data):
137
"""Return unused bytes to the medium, because they belong to the next
140
This sets the _push_back_buffer to the given bytes.
142
if not isinstance(data, bytes):
143
raise TypeError(data)
144
if self._push_back_buffer is not None:
145
raise AssertionError(
146
"_push_back called when self._push_back_buffer is %r"
147
% (self._push_back_buffer,))
150
self._push_back_buffer = data
152
def _get_push_back_buffer(self):
153
if self._push_back_buffer == b'':
154
raise AssertionError(
155
'%s._push_back_buffer should never be the empty string, '
156
'which can be confused with EOF' % (self,))
157
bytes = self._push_back_buffer
158
self._push_back_buffer = None
161
def read_bytes(self, desired_count):
162
"""Read some bytes from this medium.
164
:returns: some bytes, possibly more or less than the number requested
165
in 'desired_count' depending on the medium.
167
if self._push_back_buffer is not None:
168
return self._get_push_back_buffer()
169
bytes_to_read = min(desired_count, _MAX_READ_SIZE)
170
return self._read_bytes(bytes_to_read)
172
def _read_bytes(self, count):
173
raise NotImplementedError(self._read_bytes)
176
"""Read bytes from this request's response until a newline byte.
178
This isn't particularly efficient, so should only be used when the
179
expected size of the line is quite short.
181
:returns: a string of bytes ending in a newline (byte 0x0A).
183
line, excess = _get_line(self.read_bytes)
184
self._push_back(excess)
187
def _report_activity(self, bytes, direction):
188
"""Notify that this medium has activity.
190
Implementations should call this from all methods that actually do IO.
191
Be careful that it's not called twice, if one method is implemented on
194
:param bytes: Number of bytes read or written.
195
:param direction: 'read' or 'write' or None.
197
ui.ui_factory.report_transport_activity(self, bytes, direction)
200
_bad_file_descriptor = (errno.EBADF,)
201
if sys.platform == 'win32':
202
# Given on Windows if you pass a closed socket to select.select. Probably
203
# also given if you pass a file handle to select.
205
_bad_file_descriptor += (WSAENOTSOCK,)
208
class SmartServerStreamMedium(SmartMedium):
35
from bzrlib.smart.protocol import (
37
SmartServerRequestProtocolOne,
38
SmartServerRequestProtocolTwo,
42
from bzrlib.transport import ssh
43
except errors.ParamikoNotPresent:
44
# no paramiko. SmartSSHClientMedium will break.
48
class SmartServerStreamMedium(object):
209
49
"""Handles smart commands coming over a stream.
211
51
The stream may be a pipe connected to sshd, or a tcp socket, or an
293
90
:returns: a SmartServerRequestProtocol.
295
self._wait_for_bytes_with_timeout(self._client_timeout)
297
# We're stopping, so don't try to do any more work
92
# Identify the protocol version.
299
93
bytes = self._get_line()
300
protocol_factory, unused_bytes = _get_protocol_factory_for_bytes(bytes)
301
protocol = protocol_factory(
302
self.backing_transport, self._write_out, self.root_client_path)
303
protocol.accept_bytes(unused_bytes)
94
if bytes.startswith(REQUEST_VERSION_TWO):
95
protocol_class = SmartServerRequestProtocolTwo
96
bytes = bytes[len(REQUEST_VERSION_TWO):]
98
protocol_class = SmartServerRequestProtocolOne
99
protocol = protocol_class(self.backing_transport, self._write_out)
100
protocol.accept_bytes(bytes)
306
def _wait_on_descriptor(self, fd, timeout_seconds):
307
"""select() on a file descriptor, waiting for nonblocking read()
309
This will raise a ConnectionTimeout exception if we do not get a
310
readable handle before timeout_seconds.
313
t_end = self._timer() + timeout_seconds
314
poll_timeout = min(timeout_seconds, self._client_poll_timeout)
316
while not rs and not xs and self._timer() < t_end:
320
rs, _, xs = select.select([fd], [], [fd], poll_timeout)
321
except (select.error, socket.error) as e:
322
err = getattr(e, 'errno', None)
323
if err is None and getattr(e, 'args', None) is not None:
324
# select.error doesn't have 'errno', it just has args[0]
326
if err in _bad_file_descriptor:
327
return # Not a socket indicates read() will fail
328
elif err == errno.EINTR:
329
# Interrupted, keep looping.
333
return # Socket may already be closed
336
raise errors.ConnectionTimeout('disconnecting client after %.1f seconds'
337
% (timeout_seconds,))
339
103
def _serve_one_request(self, protocol):
340
104
"""Read one request from input, process, send back a response.
342
106
:param protocol: a SmartServerRequestProtocol.
347
109
self._serve_one_request_unguarded(protocol)
348
110
except KeyboardInterrupt:
350
except Exception as e:
351
113
self.terminate_due_to_error()
353
115
def terminate_due_to_error(self):
354
116
"""Called when an unhandled exception from the protocol occurs."""
355
117
raise NotImplementedError(self.terminate_due_to_error)
357
def _read_bytes(self, desired_count):
119
def _get_bytes(self, desired_count):
358
120
"""Get some bytes from the medium.
360
122
:param desired_count: number of bytes we want to read.
362
raise NotImplementedError(self._read_bytes)
124
raise NotImplementedError(self._get_bytes)
127
"""Read bytes from this request's response until a newline byte.
129
This isn't particularly efficient, so should only be used when the
130
expected size of the line is quite short.
132
:returns: a string of bytes ending in a newline (byte 0x0A).
134
# XXX: this duplicates SmartClientRequestProtocolOne._recv_tuple
136
while not line or line[-1] != '\n':
137
new_char = self._get_bytes(1)
140
# Ran out of bytes before receiving a complete line.
365
145
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
367
def __init__(self, sock, backing_transport, root_client_path='/',
147
def __init__(self, sock, backing_transport):
371
150
:param sock: the socket the server will read from. It will be put
372
151
into blocking mode.
374
SmartServerStreamMedium.__init__(
375
self, backing_transport, root_client_path=root_client_path,
153
SmartServerStreamMedium.__init__(self, backing_transport)
377
155
sock.setblocking(True)
378
156
self.socket = sock
379
# Get the getpeername now, as we might be closed later when we care.
381
self._client_info = sock.getpeername()
383
self._client_info = '<unknown>'
386
return '%s(client=%s)' % (self.__class__.__name__, self._client_info)
389
return '%s.%s(client=%s)' % (self.__module__, self.__class__.__name__,
392
158
def _serve_one_request_unguarded(self, protocol):
393
159
while protocol.next_read_size():
394
# We can safely try to read large chunks. If there is less data
395
# than MAX_SOCKET_CHUNK ready, the socket will just return a
396
# short read immediately rather than block.
397
bytes = self.read_bytes(osutils.MAX_SOCKET_CHUNK)
401
protocol.accept_bytes(bytes)
403
self._push_back(protocol.unused_data)
405
def _disconnect_client(self):
406
"""Close the current connection. We stopped due to a timeout/etc."""
409
def _wait_for_bytes_with_timeout(self, timeout_seconds):
410
"""Wait for more bytes to be read, but timeout if none available.
412
This allows us to detect idle connections, and stop trying to read from
413
them, without setting the socket itself to non-blocking. This also
414
allows us to specify when we watch for idle timeouts.
416
:return: None, this will raise ConnectionTimeout if we time out before
419
return self._wait_on_descriptor(self.socket, timeout_seconds)
421
def _read_bytes(self, desired_count):
422
return osutils.read_bytes_from_socket(
423
self.socket, self._report_activity)
161
protocol.accept_bytes(self.push_back)
164
bytes = self._get_bytes(4096)
168
protocol.accept_bytes(bytes)
170
self.push_back = protocol.excess_buffer
172
def _get_bytes(self, desired_count):
173
# We ignore the desired_count because on sockets it's more efficient to
175
return self.socket.recv(4096)
425
177
def terminate_due_to_error(self):
178
"""Called when an unhandled exception from the protocol occurs."""
426
179
# TODO: This should log to a server log file, but no such thing
427
180
# exists yet. Andrew Bennetts 2006-09-29.
428
181
self.socket.close()
429
182
self.finished = True
431
184
def _write_out(self, bytes):
432
tstart = osutils.timer_func()
433
osutils.send_all(self.socket, bytes, self._report_activity)
434
if 'hpss' in debug.debug_flags:
435
thread_id = thread.get_ident()
436
trace.mutter('%12s: [%s] %d bytes to the socket in %.3fs'
437
% ('wrote', thread_id, len(bytes),
438
osutils.timer_func() - tstart))
185
self.socket.sendall(bytes)
441
188
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
443
def __init__(self, in_file, out_file, backing_transport, timeout=None):
190
def __init__(self, in_file, out_file, backing_transport):
444
191
"""Construct new server.
446
193
:param in_file: Python file from which requests can be read.
447
194
:param out_file: Python file to write responses.
448
195
:param backing_transport: Transport for the directory served.
450
SmartServerStreamMedium.__init__(self, backing_transport,
197
SmartServerStreamMedium.__init__(self, backing_transport)
452
198
if sys.platform == 'win32':
453
199
# force binary mode for files
632
340
return self._read_bytes(count)
634
342
def _read_bytes(self, count):
635
"""Helper for SmartClientMediumRequest.read_bytes.
343
"""Helper for read_bytes.
637
345
read_bytes checks the state of the request to determing if bytes
638
346
should be read. After that it hands off to _read_bytes to do the
641
By default this forwards to self._medium.read_bytes because we are
642
operating on the medium's stream.
644
return self._medium.read_bytes(count)
349
raise NotImplementedError(self._read_bytes)
646
351
def read_line(self):
647
line = self._read_line()
648
if not line.endswith(b'\n'):
649
# end of file encountered reading from server
650
raise errors.ConnectionReset(
651
"Unexpected end of message. Please check connectivity "
652
"and permissions, and report a bug if problems persist.")
352
"""Read bytes from this request's response until a newline byte.
354
This isn't particularly efficient, so should only be used when the
355
expected size of the line is quite short.
357
:returns: a string of bytes ending in a newline (byte 0x0A).
359
# XXX: this duplicates SmartClientRequestProtocolOne._recv_tuple
361
while not line or line[-1] != '\n':
362
new_char = self.read_bytes(1)
365
raise errors.SmartProtocolError(
366
'unexpected end of file reading from server')
655
def _read_line(self):
656
"""Helper for SmartClientMediumRequest.read_line.
658
By default this forwards to self._medium._get_line because we are
659
operating on the medium's stream.
661
return self._medium._get_line()
664
class _VfsRefuser(object):
665
"""An object that refuses all VFS requests.
670
client._SmartClient.hooks.install_named_hook(
671
'call', self.check_vfs, 'vfs refuser')
673
def check_vfs(self, params):
675
request_method = request.request_handlers.get(params.method)
677
# A method we don't know about doesn't count as a VFS method.
679
if issubclass(request_method, vfs.VfsRequest):
680
raise HpssVfsRequestNotAllowed(params.method, params.args)
683
class _DebugCounter(object):
684
"""An object that counts the HPSS calls made to each client medium.
686
When a medium is garbage-collected, or failing that when
687
breezy.global_state exits, the total number of calls made on that medium
688
are reported via trace.note.
692
self.counts = weakref.WeakKeyDictionary()
693
client._SmartClient.hooks.install_named_hook(
694
'call', self.increment_call_count, 'hpss call counter')
695
breezy.get_global_state().cleanups.add_cleanup(self.flush_all)
697
def track(self, medium):
698
"""Start tracking calls made to a medium.
700
This only keeps a weakref to the medium, so shouldn't affect the
703
medium_repr = repr(medium)
704
# Add this medium to the WeakKeyDictionary
705
self.counts[medium] = dict(count=0, vfs_count=0,
706
medium_repr=medium_repr)
707
# Weakref callbacks are fired in reverse order of their association
708
# with the referenced object. So we add a weakref *after* adding to
709
# the WeakKeyDict so that we can report the value from it before the
710
# entry is removed by the WeakKeyDict's own callback.
711
ref = weakref.ref(medium, self.done)
713
def increment_call_count(self, params):
714
# Increment the count in the WeakKeyDictionary
715
value = self.counts[params.medium]
718
request_method = request.request_handlers.get(params.method)
720
# A method we don't know about doesn't count as a VFS method.
722
if issubclass(request_method, vfs.VfsRequest):
723
value['vfs_count'] += 1
726
value = self.counts[ref]
727
count, vfs_count, medium_repr = (
728
value['count'], value['vfs_count'], value['medium_repr'])
729
# In case this callback is invoked for the same ref twice (by the
730
# weakref callback and by the atexit function), set the call count back
731
# to 0 so this item won't be reported twice.
733
value['vfs_count'] = 0
735
trace.note(gettext('HPSS calls: {0} ({1} vfs) {2}').format(
736
count, vfs_count, medium_repr))
739
for ref in list(self.counts.keys()):
742
_debug_counter = None
746
class SmartClientMedium(SmartMedium):
370
class SmartClientMedium(object):
747
371
"""Smart client is a medium for sending smart protocol requests over."""
749
def __init__(self, base):
750
super(SmartClientMedium, self).__init__()
752
self._protocol_version_error = None
753
self._protocol_version = None
754
self._done_hello = False
755
# Be optimistic: we assume the remote end can accept new remote
756
# requests until we get an error saying otherwise.
757
# _remote_version_is_before tracks the bzr version the remote side
758
# can be based on what we've seen so far.
759
self._remote_version_is_before = None
760
# Install debug hook function if debug flag is set.
761
if 'hpss' in debug.debug_flags:
762
global _debug_counter
763
if _debug_counter is None:
764
_debug_counter = _DebugCounter()
765
_debug_counter.track(self)
766
if 'hpss_client_no_vfs' in debug.debug_flags:
768
if _vfs_refuser is None:
769
_vfs_refuser = _VfsRefuser()
771
def _is_remote_before(self, version_tuple):
772
"""Is it possible the remote side supports RPCs for a given version?
776
needed_version = (1, 2)
777
if medium._is_remote_before(needed_version):
778
fallback_to_pre_1_2_rpc()
782
except UnknownSmartMethod:
783
medium._remember_remote_is_before(needed_version)
784
fallback_to_pre_1_2_rpc()
786
:seealso: _remember_remote_is_before
788
if self._remote_version_is_before is None:
789
# So far, the remote side seems to support everything
791
return version_tuple >= self._remote_version_is_before
793
def _remember_remote_is_before(self, version_tuple):
794
"""Tell this medium that the remote side is older the given version.
796
:seealso: _is_remote_before
798
if (self._remote_version_is_before is not None and
799
version_tuple > self._remote_version_is_before):
800
# We have been told that the remote side is older than some version
801
# which is newer than a previously supplied older-than version.
802
# This indicates that some smart verb call is not guarded
803
# appropriately (it should simply not have been tried).
805
"_remember_remote_is_before(%r) called, but "
806
"_remember_remote_is_before(%r) was called previously."
807
, version_tuple, self._remote_version_is_before)
808
if 'hpss' in debug.debug_flags:
809
ui.ui_factory.show_warning(
810
"_remember_remote_is_before(%r) called, but "
811
"_remember_remote_is_before(%r) was called previously."
812
% (version_tuple, self._remote_version_is_before))
814
self._remote_version_is_before = version_tuple
816
def protocol_version(self):
817
"""Find out if 'hello' smart request works."""
818
if self._protocol_version_error is not None:
819
raise self._protocol_version_error
820
if not self._done_hello:
822
medium_request = self.get_request()
823
# Send a 'hello' request in protocol version one, for maximum
824
# backwards compatibility.
825
client_protocol = protocol.SmartClientRequestProtocolOne(medium_request)
826
client_protocol.query_version()
827
self._done_hello = True
828
except errors.SmartProtocolError as e:
829
# Cache the error, just like we would cache a successful
831
self._protocol_version_error = e
835
def should_probe(self):
836
"""Should RemoteBzrDirFormat.probe_transport send a smart request on
839
Some transports are unambiguously smart-only; there's no need to check
840
if the transport is able to carry smart requests, because that's all
841
it is for. In those cases, this method should return False.
843
But some HTTP transports can sometimes fail to carry smart requests,
844
but still be usuable for accessing remote bzrdirs via plain file
845
accesses. So for those transports, their media should return True here
846
so that RemoteBzrDirFormat can determine if it is appropriate for that
851
373
def disconnect(self):
852
374
"""If this medium maintains a persistent connection, close it.
854
376
The default implementation does nothing.
857
def remote_path_from_transport(self, transport):
858
"""Convert transport into a path suitable for using in a request.
860
Note that the resulting remote path doesn't encode the host name or
861
anything but path, so it is only safe to use it in requests sent over
862
the medium from the matching transport.
864
medium_base = urlutils.join(self.base, '/')
865
rel_url = urlutils.relative_url(medium_base, transport.base)
866
return urlutils.unquote(rel_url)
869
380
class SmartClientStreamMedium(SmartClientMedium):
870
381
"""Stream based medium common class.
905
415
return SmartClientStreamMediumRequest(self)
908
"""We have been disconnected, reset current state.
910
This resets things like _current_request and connected state.
913
self._current_request = None
417
def read_bytes(self, count):
418
return self._read_bytes(count)
916
421
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
917
422
"""A client medium using simple pipes.
919
424
This client does not manage the pipes: it assumes they will always be open.
922
def __init__(self, readable_pipe, writeable_pipe, base):
923
SmartClientStreamMedium.__init__(self, base)
427
def __init__(self, readable_pipe, writeable_pipe):
428
SmartClientStreamMedium.__init__(self)
924
429
self._readable_pipe = readable_pipe
925
430
self._writeable_pipe = writeable_pipe
927
def _accept_bytes(self, data):
432
def _accept_bytes(self, bytes):
928
433
"""See SmartClientStreamMedium.accept_bytes."""
930
self._writeable_pipe.write(data)
932
if e.errno in (errno.EINVAL, errno.EPIPE):
933
raise errors.ConnectionReset(
934
"Error trying to write to subprocess", e)
936
self._report_activity(len(data), 'write')
434
self._writeable_pipe.write(bytes)
938
436
def _flush(self):
939
437
"""See SmartClientStreamMedium._flush()."""
940
# Note: If flush were to fail, we'd like to raise ConnectionReset, etc.
941
# However, testing shows that even when the child process is
942
# gone, this doesn't error.
943
438
self._writeable_pipe.flush()
945
440
def _read_bytes(self, count):
946
441
"""See SmartClientStreamMedium._read_bytes."""
947
bytes_to_read = min(count, _MAX_READ_SIZE)
948
data = self._readable_pipe.read(bytes_to_read)
949
self._report_activity(len(data), 'read')
953
class SSHParams(object):
954
"""A set of parameters for starting a remote bzr via SSH."""
442
return self._readable_pipe.read(count)
445
class SmartSSHClientMedium(SmartClientStreamMedium):
446
"""A client medium using SSH."""
956
448
def __init__(self, host, port=None, username=None, password=None,
957
bzr_remote_path='bzr'):
960
self.username = username
961
self.password = password
962
self.bzr_remote_path = bzr_remote_path
965
class SmartSSHClientMedium(SmartClientStreamMedium):
966
"""A client medium using SSH.
968
It delegates IO to a SmartSimplePipesClientMedium or
969
SmartClientAlreadyConnectedSocketMedium (depending on platform).
972
def __init__(self, base, ssh_params, vendor=None):
449
vendor=None, bzr_remote_path=None):
973
450
"""Creates a client that will connect on the first use.
975
:param ssh_params: A SSHParams instance.
976
452
:param vendor: An optional override for the ssh vendor to use. See
977
breezy.transport.ssh for details on ssh vendors.
453
bzrlib.transport.ssh for details on ssh vendors.
979
self._real_medium = None
980
self._ssh_params = ssh_params
981
# for the benefit of progress making a short description of this
983
self._scheme = 'bzr+ssh'
984
# SmartClientStreamMedium stores the repr of this object in its
985
# _DebugCounter so we have to store all the values used in our repr
986
# method before calling the super init.
987
SmartClientStreamMedium.__init__(self, base)
455
SmartClientStreamMedium.__init__(self)
456
self._connected = False
458
self._password = password
460
self._username = username
461
self._read_from = None
462
self._ssh_connection = None
988
463
self._vendor = vendor
989
self._ssh_connection = None
992
if self._ssh_params.port is None:
995
maybe_port = ':%s' % self._ssh_params.port
996
if self._ssh_params.username is None:
999
maybe_user = '%s@' % self._ssh_params.username
1000
return "%s(%s://%s%s%s/)" % (
1001
self.__class__.__name__,
1004
self._ssh_params.host,
464
self._write_to = None
465
self._bzr_remote_path = bzr_remote_path
466
if self._bzr_remote_path is None:
467
symbol_versioning.warn(
468
'bzr_remote_path is required as of bzr 0.92',
469
DeprecationWarning, stacklevel=2)
470
self._bzr_remote_path = os.environ.get('BZR_REMOTE_PATH', 'bzr')
1007
472
def _accept_bytes(self, bytes):
1008
473
"""See SmartClientStreamMedium.accept_bytes."""
1009
474
self._ensure_connection()
1010
self._real_medium.accept_bytes(bytes)
475
self._write_to.write(bytes)
1012
477
def disconnect(self):
1013
478
"""See SmartClientMedium.disconnect()."""
1014
if self._real_medium is not None:
1015
self._real_medium.disconnect()
1016
self._real_medium = None
1017
if self._ssh_connection is not None:
1018
self._ssh_connection.close()
1019
self._ssh_connection = None
479
if not self._connected:
481
self._read_from.close()
482
self._write_to.close()
483
self._ssh_connection.close()
484
self._connected = False
1021
486
def _ensure_connection(self):
1022
487
"""Connect this medium if not already connected."""
1023
if self._real_medium is not None:
1025
490
if self._vendor is None:
1026
491
vendor = ssh._get_ssh_vendor()
1028
493
vendor = self._vendor
1029
self._ssh_connection = vendor.connect_ssh(self._ssh_params.username,
1030
self._ssh_params.password, self._ssh_params.host,
1031
self._ssh_params.port,
1032
command=[self._ssh_params.bzr_remote_path, 'serve', '--inet',
494
self._ssh_connection = vendor.connect_ssh(self._username,
495
self._password, self._host, self._port,
496
command=[self._bzr_remote_path, 'serve', '--inet',
1033
497
'--directory=/', '--allow-writes'])
1034
io_kind, io_object = self._ssh_connection.get_sock_or_pipes()
1035
if io_kind == 'socket':
1036
self._real_medium = SmartClientAlreadyConnectedSocketMedium(
1037
self.base, io_object)
1038
elif io_kind == 'pipes':
1039
read_from, write_to = io_object
1040
self._real_medium = SmartSimplePipesClientMedium(
1041
read_from, write_to, self.base)
1043
raise AssertionError(
1044
"Unexpected io_kind %r from %r"
1045
% (io_kind, self._ssh_connection))
1046
for hook in transport.Transport.hooks["post_connect"]:
498
self._read_from, self._write_to = \
499
self._ssh_connection.get_filelike_channels()
500
self._connected = True
1049
502
def _flush(self):
1050
503
"""See SmartClientStreamMedium._flush()."""
1051
self._real_medium._flush()
504
self._write_to.flush()
1053
506
def _read_bytes(self, count):
1054
507
"""See SmartClientStreamMedium.read_bytes."""
1055
if self._real_medium is None:
508
if not self._connected:
1056
509
raise errors.MediumNotConnected(self)
1057
return self._real_medium.read_bytes(count)
1060
# Port 4155 is the default port for bzr://, registered with IANA.
1061
BZR_DEFAULT_INTERFACE = None
1062
BZR_DEFAULT_PORT = 4155
1065
class SmartClientSocketMedium(SmartClientStreamMedium):
1066
"""A client medium using a socket.
1068
This class isn't usable directly. Use one of its subclasses instead.
1071
def __init__(self, base):
1072
SmartClientStreamMedium.__init__(self, base)
510
return self._read_from.read(count)
513
class SmartTCPClientMedium(SmartClientStreamMedium):
514
"""A client medium using TCP."""
516
def __init__(self, host, port):
517
"""Creates a client that will connect on the first use."""
518
SmartClientStreamMedium.__init__(self)
519
self._connected = False
1073
522
self._socket = None
1074
self._connected = False
1076
524
def _accept_bytes(self, bytes):
1077
525
"""See SmartClientMedium.accept_bytes."""
1078
526
self._ensure_connection()
1079
osutils.send_all(self._socket, bytes, self._report_activity)
527
self._socket.sendall(bytes)
529
def disconnect(self):
530
"""See SmartClientMedium.disconnect()."""
531
if not self._connected:
535
self._connected = False
1081
537
def _ensure_connection(self):
1082
538
"""Connect this medium if not already connected."""
1083
raise NotImplementedError(self._ensure_connection)
541
self._socket = socket.socket()
542
self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
543
result = self._socket.connect_ex((self._host, int(self._port)))
545
raise errors.ConnectionError("failed to connect to %s:%d: %s" %
546
(self._host, self._port, os.strerror(result)))
547
self._connected = True
1085
549
def _flush(self):
1086
550
"""See SmartClientStreamMedium._flush().
1088
For sockets we do no flushing. For TCP sockets we may want to turn off
1089
TCP_NODELAY and add a means to do a flush, but that can be done in the
552
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
553
add a means to do a flush, but that can be done in the future.
1093
556
def _read_bytes(self, count):
1094
557
"""See SmartClientMedium.read_bytes."""
1095
558
if not self._connected:
1096
559
raise errors.MediumNotConnected(self)
1097
return osutils.read_bytes_from_socket(
1098
self._socket, self._report_activity)
1100
def disconnect(self):
1101
"""See SmartClientMedium.disconnect()."""
1102
if not self._connected:
1104
self._socket.close()
1106
self._connected = False
1109
class SmartTCPClientMedium(SmartClientSocketMedium):
1110
"""A client medium that creates a TCP connection."""
1112
def __init__(self, host, port, base):
1113
"""Creates a client that will connect on the first use."""
1114
SmartClientSocketMedium.__init__(self, base)
1118
def _ensure_connection(self):
1119
"""Connect this medium if not already connected."""
1122
if self._port is None:
1123
port = BZR_DEFAULT_PORT
1125
port = int(self._port)
1127
sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC,
1128
socket.SOCK_STREAM, 0, 0)
1129
except socket.gaierror as xxx_todo_changeme:
1130
(err_num, err_msg) = xxx_todo_changeme.args
1131
raise errors.ConnectionError("failed to lookup %s:%d: %s" %
1132
(self._host, port, err_msg))
1133
# Initialize err in case there are no addresses returned:
1134
last_err = socket.error("no address found for %s" % self._host)
1135
for (family, socktype, proto, canonname, sockaddr) in sockaddrs:
1137
self._socket = socket.socket(family, socktype, proto)
1138
self._socket.setsockopt(socket.IPPROTO_TCP,
1139
socket.TCP_NODELAY, 1)
1140
self._socket.connect(sockaddr)
1141
except socket.error as err:
1142
if self._socket is not None:
1143
self._socket.close()
1148
if self._socket is None:
1149
# socket errors either have a (string) or (errno, string) as their
1151
if isinstance(last_err.args, str):
1152
err_msg = last_err.args
1154
err_msg = last_err.args[1]
1155
raise errors.ConnectionError("failed to connect to %s:%d: %s" %
1156
(self._host, port, err_msg))
1157
self._connected = True
1158
for hook in transport.Transport.hooks["post_connect"]:
1162
class SmartClientAlreadyConnectedSocketMedium(SmartClientSocketMedium):
1163
"""A client medium for an already connected socket.
1165
Note that this class will assume it "owns" the socket, so it will close it
1166
when its disconnect method is called.
1169
def __init__(self, base, sock):
1170
SmartClientSocketMedium.__init__(self, base)
1172
self._connected = True
1174
def _ensure_connection(self):
1175
# Already connected, by definition! So nothing to do.
560
return self._socket.recv(count)
1179
563
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
added
removed
bzrlib/smart/medium.py
