45
46
from bzrlib.smart import client, protocol, request, vfs
46
47
from bzrlib.transport import ssh
48
#usually already imported, and getting IllegalScoperReplacer on it here.
49
49
from bzrlib import osutils
51
# We must not read any more than 64k at a time so we don't risk "no buffer
52
# space available" errors on some platforms. Windows in particular is likely
53
# to give error 10053 or 10055 if we read more than 64k from a socket.
54
_MAX_READ_SIZE = 64 * 1024
51
# Throughout this module buffer size parameters are either limited to be at
52
# most _MAX_READ_SIZE, or are ignored and _MAX_READ_SIZE is used instead.
53
# For this module's purposes, MAX_SOCKET_CHUNK is a reasonable size for reads
54
# from non-sockets as well.
55
_MAX_READ_SIZE = osutils.MAX_SOCKET_CHUNK
57
57
def _get_protocol_factory_for_bytes(bytes):
58
58
"""Determine the right protocol factory for 'bytes'.
722
734
def _read_bytes(self, count):
723
735
"""See SmartClientStreamMedium._read_bytes."""
724
bytes = self._readable_pipe.read(count)
736
bytes_to_read = min(count, _MAX_READ_SIZE)
737
bytes = self._readable_pipe.read(bytes_to_read)
725
738
self._report_activity(len(bytes), 'read')
742
class SSHParams(object):
743
"""A set of parameters for starting a remote bzr via SSH."""
745
def __init__(self, host, port=None, username=None, password=None,
746
bzr_remote_path='bzr'):
749
self.username = username
750
self.password = password
751
self.bzr_remote_path = bzr_remote_path
729
754
class SmartSSHClientMedium(SmartClientStreamMedium):
730
"""A client medium using SSH."""
755
"""A client medium using SSH.
757
It delegates IO to a SmartClientSocketMedium or
758
SmartClientAlreadyConnectedSocketMedium (depending on platform).
732
def __init__(self, host, port=None, username=None, password=None,
733
base=None, vendor=None, bzr_remote_path=None):
761
def __init__(self, base, ssh_params, vendor=None):
734
762
"""Creates a client that will connect on the first use.
764
:param ssh_params: A SSHParams instance.
736
765
:param vendor: An optional override for the ssh vendor to use. See
737
766
bzrlib.transport.ssh for details on ssh vendors.
739
self._connected = False
741
self._password = password
743
self._username = username
768
self._real_medium = None
769
self._ssh_params = ssh_params
770
# for the benefit of progress making a short description of this
772
self._scheme = 'bzr+ssh'
744
773
# SmartClientStreamMedium stores the repr of this object in its
745
774
# _DebugCounter so we have to store all the values used in our repr
746
775
# method before calling the super init.
747
776
SmartClientStreamMedium.__init__(self, base)
748
self._read_from = None
777
self._vendor = vendor
749
778
self._ssh_connection = None
750
self._vendor = vendor
751
self._write_to = None
752
self._bzr_remote_path = bzr_remote_path
753
# for the benefit of progress making a short description of this
755
self._scheme = 'bzr+ssh'
757
780
def __repr__(self):
758
return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
781
if self._ssh_params.port is None:
784
maybe_port = ':%s' % self._ssh_params.port
785
return "%s(%s://%s@%s%s/)" % (
759
786
self.__class__.__name__,
788
self._ssh_params.username,
789
self._ssh_params.host,
765
792
def _accept_bytes(self, bytes):
766
793
"""See SmartClientStreamMedium.accept_bytes."""
767
794
self._ensure_connection()
768
self._write_to.write(bytes)
769
self._report_activity(len(bytes), 'write')
795
self._real_medium.accept_bytes(bytes)
771
797
def disconnect(self):
772
798
"""See SmartClientMedium.disconnect()."""
773
if not self._connected:
775
self._read_from.close()
776
self._write_to.close()
777
self._ssh_connection.close()
778
self._connected = False
799
if self._real_medium is not None:
800
self._real_medium.disconnect()
801
self._real_medium = None
802
if self._ssh_connection is not None:
803
self._ssh_connection.close()
804
self._ssh_connection = None
780
806
def _ensure_connection(self):
781
807
"""Connect this medium if not already connected."""
808
if self._real_medium is not None:
784
810
if self._vendor is None:
785
811
vendor = ssh._get_ssh_vendor()
787
813
vendor = self._vendor
788
self._ssh_connection = vendor.connect_ssh(self._username,
789
self._password, self._host, self._port,
790
command=[self._bzr_remote_path, 'serve', '--inet',
814
self._ssh_connection = vendor.connect_ssh(self._ssh_params.username,
815
self._ssh_params.password, self._ssh_params.host,
816
self._ssh_params.port,
817
command=[self._ssh_params.bzr_remote_path, 'serve', '--inet',
791
818
'--directory=/', '--allow-writes'])
792
self._read_from, self._write_to = \
793
self._ssh_connection.get_filelike_channels()
794
self._connected = True
819
io_kind, io_object = self._ssh_connection.get_sock_or_pipes()
820
if io_kind == 'socket':
821
self._real_medium = SmartClientAlreadyConnectedSocketMedium(
822
self.base, io_object)
823
elif io_kind == 'pipes':
824
read_from, write_to = io_object
825
self._real_medium = SmartSimplePipesClientMedium(
826
read_from, write_to, self.base)
828
raise AssertionError(
829
"Unexpected io_kind %r from %r"
830
% (io_kind, self._ssh_connection))
796
832
def _flush(self):
797
833
"""See SmartClientStreamMedium._flush()."""
798
self._write_to.flush()
834
self._real_medium._flush()
800
836
def _read_bytes(self, count):
801
837
"""See SmartClientStreamMedium.read_bytes."""
802
if not self._connected:
838
if self._real_medium is None:
803
839
raise errors.MediumNotConnected(self)
804
bytes_to_read = min(count, _MAX_READ_SIZE)
805
bytes = self._read_from.read(bytes_to_read)
806
self._report_activity(len(bytes), 'read')
840
return self._real_medium.read_bytes(count)
810
843
# Port 4155 is the default port for bzr://, registered with IANA.
812
845
BZR_DEFAULT_PORT = 4155
815
class SmartTCPClientMedium(SmartClientStreamMedium):
816
"""A client medium using TCP."""
848
class SmartClientSocketMedium(SmartClientStreamMedium):
849
"""A client medium using a socket.
851
This class isn't usable directly. Use one of its subclasses instead.
818
def __init__(self, host, port, base):
819
"""Creates a client that will connect on the first use."""
854
def __init__(self, base):
820
855
SmartClientStreamMedium.__init__(self, base)
821
857
self._connected = False
826
859
def _accept_bytes(self, bytes):
827
860
"""See SmartClientMedium.accept_bytes."""
828
861
self._ensure_connection()
829
862
osutils.send_all(self._socket, bytes, self._report_activity)
864
def _ensure_connection(self):
865
"""Connect this medium if not already connected."""
866
raise NotImplementedError(self._ensure_connection)
869
"""See SmartClientStreamMedium._flush().
871
For sockets we do no flushing. For TCP sockets we may want to turn off
872
TCP_NODELAY and add a means to do a flush, but that can be done in the
876
def _read_bytes(self, count):
877
"""See SmartClientMedium.read_bytes."""
878
if not self._connected:
879
raise errors.MediumNotConnected(self)
880
return osutils.read_bytes_from_socket(
881
self._socket, self._report_activity)
831
883
def disconnect(self):
832
884
"""See SmartClientMedium.disconnect()."""
833
885
if not self._connected:
875
937
(self._host, port, err_msg))
876
938
self._connected = True
879
"""See SmartClientStreamMedium._flush().
881
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
882
add a means to do a flush, but that can be done in the future.
885
def _read_bytes(self, count):
886
"""See SmartClientMedium.read_bytes."""
887
if not self._connected:
888
raise errors.MediumNotConnected(self)
889
return _read_bytes_from_socket(
890
self._socket.recv, count, self._report_activity)
941
class SmartClientAlreadyConnectedSocketMedium(SmartClientSocketMedium):
942
"""A client medium for an already connected socket.
944
Note that this class will assume it "owns" the socket, so it will close it
945
when its disconnect method is called.
948
def __init__(self, base, sock):
949
SmartClientSocketMedium.__init__(self, base)
951
self._connected = True
953
def _ensure_connection(self):
954
# Already connected, by definition! So nothing to do.
893
958
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
added
removed
bzrlib/smart/medium.py
