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):
30
from bzrlib import errors
31
from bzrlib.smart import protocol
33
from bzrlib.transport import ssh
34
except errors.ParamikoNotPresent:
35
# no paramiko. SmartSSHClientMedium will break.
39
class SmartServerStreamMedium(object):
209
40
"""Handles smart commands coming over a stream.
211
42
The stream may be a pipe connected to sshd, or a tcp socket, or an
247
65
from sys import stderr
249
67
while not self.finished:
250
server_protocol = self._build_protocol()
68
server_protocol = protocol.SmartServerRequestProtocolOne(
69
self.backing_transport, self._write_out)
251
70
self._serve_one_request(server_protocol)
252
except errors.ConnectionTimeout as e:
253
trace.note('%s' % (e,))
254
trace.log_exception_quietly()
255
self._disconnect_client()
256
# We reported it, no reason to make a big fuss.
258
except Exception as e:
259
72
stderr.write("%s terminating on exception %s\n" % (self, e))
261
self._disconnect_client()
263
def _stop_gracefully(self):
264
"""When we finish this message, stop looking for more."""
265
trace.mutter('Stopping %s' % (self,))
268
def _disconnect_client(self):
269
"""Close the current connection. We stopped due to a timeout/etc."""
270
# The default implementation is a no-op, because that is all we used to
271
# do when disconnecting from a client. I suppose we never had the
272
# *server* initiate a disconnect, before
274
def _wait_for_bytes_with_timeout(self, timeout_seconds):
275
"""Wait for more bytes to be read, but timeout if none available.
277
This allows us to detect idle connections, and stop trying to read from
278
them, without setting the socket itself to non-blocking. This also
279
allows us to specify when we watch for idle timeouts.
281
:return: Did we timeout? (True if we timed out, False if there is data
284
raise NotImplementedError(self._wait_for_bytes_with_timeout)
286
def _build_protocol(self):
287
"""Identifies the version of the incoming request, and returns an
288
a protocol object that can interpret it.
290
If more bytes than the version prefix of the request are read, they will
291
be fed into the protocol before it is returned.
293
: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
299
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)
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
75
def _serve_one_request(self, protocol):
340
76
"""Read one request from input, process, send back a response.
342
78
:param protocol: a SmartServerRequestProtocol.
347
81
self._serve_one_request_unguarded(protocol)
348
82
except KeyboardInterrupt:
350
except Exception as e:
351
85
self.terminate_due_to_error()
353
87
def terminate_due_to_error(self):
354
88
"""Called when an unhandled exception from the protocol occurs."""
355
89
raise NotImplementedError(self.terminate_due_to_error)
357
def _read_bytes(self, desired_count):
358
"""Get some bytes from the medium.
360
:param desired_count: number of bytes we want to read.
362
raise NotImplementedError(self._read_bytes)
365
92
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
367
def __init__(self, sock, backing_transport, root_client_path='/',
94
def __init__(self, sock, backing_transport):
371
97
:param sock: the socket the server will read from. It will be put
372
98
into blocking mode.
374
SmartServerStreamMedium.__init__(
375
self, backing_transport, root_client_path=root_client_path,
100
SmartServerStreamMedium.__init__(self, backing_transport)
377
102
sock.setblocking(True)
378
103
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
105
def _serve_one_request_unguarded(self, protocol):
393
106
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)
108
protocol.accept_bytes(self.push_back)
111
bytes = self.socket.recv(4096)
115
protocol.accept_bytes(bytes)
117
self.push_back = protocol.excess_buffer
425
119
def terminate_due_to_error(self):
120
"""Called when an unhandled exception from the protocol occurs."""
426
121
# TODO: This should log to a server log file, but no such thing
427
122
# exists yet. Andrew Bennetts 2006-09-29.
428
123
self.socket.close()
429
124
self.finished = True
431
126
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))
127
self.socket.sendall(bytes)
441
130
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
443
def __init__(self, in_file, out_file, backing_transport, timeout=None):
132
def __init__(self, in_file, out_file, backing_transport):
444
133
"""Construct new server.
446
135
:param in_file: Python file from which requests can be read.
447
136
:param out_file: Python file to write responses.
448
137
:param backing_transport: Transport for the directory served.
450
SmartServerStreamMedium.__init__(self, backing_transport,
452
if sys.platform == 'win32':
453
# force binary mode for files
455
for f in (in_file, out_file):
456
fileno = getattr(f, 'fileno', None)
458
msvcrt.setmode(fileno(), os.O_BINARY)
139
SmartServerStreamMedium.__init__(self, backing_transport)
459
140
self._in = in_file
460
141
self._out = out_file
463
"""See SmartServerStreamMedium.serve"""
464
# This is the regular serve, except it adds signal trapping for soft
466
stop_gracefully = self._stop_gracefully
467
signals.register_on_hangup(id(self), stop_gracefully)
469
return super(SmartServerPipeStreamMedium, self).serve()
471
signals.unregister_on_hangup(id(self))
473
143
def _serve_one_request_unguarded(self, protocol):
475
# We need to be careful not to read past the end of the current
476
# request, or else the read from the pipe will block, so we use
477
# protocol.next_read_size().
478
145
bytes_to_read = protocol.next_read_size()
479
146
if bytes_to_read == 0:
480
147
# Finished serving this request.
481
148
self._out.flush()
483
bytes = self.read_bytes(bytes_to_read)
150
bytes = self._in.read(bytes_to_read)
485
152
# Connection has been closed.
486
153
self.finished = True
487
154
self._out.flush()
489
156
protocol.accept_bytes(bytes)
491
def _disconnect_client(self):
496
def _wait_for_bytes_with_timeout(self, timeout_seconds):
497
"""Wait for more bytes to be read, but timeout if none available.
499
This allows us to detect idle connections, and stop trying to read from
500
them, without setting the socket itself to non-blocking. This also
501
allows us to specify when we watch for idle timeouts.
503
:return: None, this will raise ConnectionTimeout if we time out before
506
if (getattr(self._in, 'fileno', None) is None
507
or sys.platform == 'win32'):
508
# You can't select() file descriptors on Windows.
511
return self._wait_on_descriptor(self._in, timeout_seconds)
512
except io.UnsupportedOperation:
515
def _read_bytes(self, desired_count):
516
return self._in.read(desired_count)
518
158
def terminate_due_to_error(self):
519
159
# TODO: This should log to a server log file, but no such thing
520
160
# exists yet. Andrew Bennetts 2006-09-29.
632
272
return self._read_bytes(count)
634
274
def _read_bytes(self, count):
635
"""Helper for SmartClientMediumRequest.read_bytes.
275
"""Helper for read_bytes.
637
277
read_bytes checks the state of the request to determing if bytes
638
278
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)
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.")
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):
281
raise NotImplementedError(self._read_bytes)
284
class SmartClientMedium(object):
747
285
"""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
287
def disconnect(self):
852
288
"""If this medium maintains a persistent connection, close it.
854
290
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
294
class SmartClientStreamMedium(SmartClientMedium):
870
295
"""Stream based medium common class.
905
329
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
331
def read_bytes(self, count):
332
return self._read_bytes(count)
916
335
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
917
336
"""A client medium using simple pipes.
919
338
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)
341
def __init__(self, readable_pipe, writeable_pipe):
342
SmartClientStreamMedium.__init__(self)
924
343
self._readable_pipe = readable_pipe
925
344
self._writeable_pipe = writeable_pipe
927
def _accept_bytes(self, data):
346
def _accept_bytes(self, bytes):
928
347
"""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')
348
self._writeable_pipe.write(bytes)
938
350
def _flush(self):
939
351
"""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
352
self._writeable_pipe.flush()
945
354
def _read_bytes(self, count):
946
355
"""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."""
356
return self._readable_pipe.read(count)
359
class SmartSSHClientMedium(SmartClientStreamMedium):
360
"""A client medium using SSH."""
956
362
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):
973
364
"""Creates a client that will connect on the first use.
975
:param ssh_params: A SSHParams instance.
976
366
:param vendor: An optional override for the ssh vendor to use. See
977
breezy.transport.ssh for details on ssh vendors.
367
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)
369
SmartClientStreamMedium.__init__(self)
370
self._connected = False
372
self._password = password
374
self._username = username
375
self._read_from = None
376
self._ssh_connection = None
988
377
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,
378
self._write_to = None
1007
380
def _accept_bytes(self, bytes):
1008
381
"""See SmartClientStreamMedium.accept_bytes."""
1009
382
self._ensure_connection()
1010
self._real_medium.accept_bytes(bytes)
383
self._write_to.write(bytes)
1012
385
def disconnect(self):
1013
386
"""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
387
if not self._connected:
389
self._read_from.close()
390
self._write_to.close()
391
self._ssh_connection.close()
392
self._connected = False
1021
394
def _ensure_connection(self):
1022
395
"""Connect this medium if not already connected."""
1023
if self._real_medium is not None:
398
executable = os.environ.get('BZR_REMOTE_PATH', 'bzr')
1025
399
if self._vendor is None:
1026
400
vendor = ssh._get_ssh_vendor()
1028
402
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',
1033
'--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"]:
403
self._ssh_connection = vendor.connect_ssh(self._username,
404
self._password, self._host, self._port,
405
command=[executable, 'serve', '--inet', '--directory=/',
407
self._read_from, self._write_to = \
408
self._ssh_connection.get_filelike_channels()
409
self._connected = True
1049
411
def _flush(self):
1050
412
"""See SmartClientStreamMedium._flush()."""
1051
self._real_medium._flush()
413
self._write_to.flush()
1053
415
def _read_bytes(self, count):
1054
416
"""See SmartClientStreamMedium.read_bytes."""
1055
if self._real_medium is None:
417
if not self._connected:
1056
418
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)
419
return self._read_from.read(count)
422
class SmartTCPClientMedium(SmartClientStreamMedium):
423
"""A client medium using TCP."""
425
def __init__(self, host, port):
426
"""Creates a client that will connect on the first use."""
427
SmartClientStreamMedium.__init__(self)
428
self._connected = False
1073
431
self._socket = None
1074
self._connected = False
1076
433
def _accept_bytes(self, bytes):
1077
434
"""See SmartClientMedium.accept_bytes."""
1078
435
self._ensure_connection()
1079
osutils.send_all(self._socket, bytes, self._report_activity)
436
self._socket.sendall(bytes)
438
def disconnect(self):
439
"""See SmartClientMedium.disconnect()."""
440
if not self._connected:
444
self._connected = False
1081
446
def _ensure_connection(self):
1082
447
"""Connect this medium if not already connected."""
1083
raise NotImplementedError(self._ensure_connection)
450
self._socket = socket.socket()
451
self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
452
result = self._socket.connect_ex((self._host, int(self._port)))
454
raise errors.ConnectionError("failed to connect to %s:%d: %s" %
455
(self._host, self._port, os.strerror(result)))
456
self._connected = True
1085
458
def _flush(self):
1086
459
"""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
461
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
462
add a means to do a flush, but that can be done in the future.
1093
465
def _read_bytes(self, count):
1094
466
"""See SmartClientMedium.read_bytes."""
1095
467
if not self._connected:
1096
468
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.
469
return self._socket.recv(count)
1179
472
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
added
removed
bzrlib/smart/medium.py
