285
287
self._push_back(protocol.unused_data)
287
289
def _read_bytes(self, desired_count):
288
return osutils.read_bytes_from_socket(
289
self.socket, self._report_activity)
290
return _read_bytes_from_socket(
291
self.socket.recv, desired_count, self._report_activity)
291
293
def terminate_due_to_error(self):
292
294
# TODO: This should log to a server log file, but no such thing
293
295
# exists yet. Andrew Bennetts 2006-09-29.
296
osutils.until_no_eintr(self.socket.close)
295
297
self.finished = True
297
299
def _write_out(self, bytes):
298
300
tstart = osutils.timer_func()
299
301
osutils.send_all(self.socket, bytes, self._report_activity)
300
302
if 'hpss' in debug.debug_flags:
301
thread_id = thread.get_ident()
303
thread_id = threading.currentThread().ident
302
304
trace.mutter('%12s: [%s] %d bytes to the socket in %.3fs'
303
305
% ('wrote', thread_id, len(bytes),
304
306
osutils.timer_func() - tstart))
725
721
def _accept_bytes(self, bytes):
726
722
"""See SmartClientStreamMedium.accept_bytes."""
727
self._writeable_pipe.write(bytes)
723
osutils.until_no_eintr(self._writeable_pipe.write, bytes)
728
724
self._report_activity(len(bytes), 'write')
730
726
def _flush(self):
731
727
"""See SmartClientStreamMedium._flush()."""
732
self._writeable_pipe.flush()
728
osutils.until_no_eintr(self._writeable_pipe.flush)
734
730
def _read_bytes(self, count):
735
731
"""See SmartClientStreamMedium._read_bytes."""
736
bytes_to_read = min(count, _MAX_READ_SIZE)
737
bytes = self._readable_pipe.read(bytes_to_read)
732
bytes = osutils.until_no_eintr(self._readable_pipe.read, count)
738
733
self._report_activity(len(bytes), 'read')
742
class SSHParams(object):
743
"""A set of parameters for starting a remote bzr via SSH."""
737
class SmartSSHClientMedium(SmartClientStreamMedium):
738
"""A client medium using SSH."""
745
740
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
754
class SmartSSHClientMedium(SmartClientStreamMedium):
755
"""A client medium using SSH.
757
It delegates IO to a SmartClientSocketMedium or
758
SmartClientAlreadyConnectedSocketMedium (depending on platform).
761
def __init__(self, base, ssh_params, vendor=None):
741
base=None, vendor=None, bzr_remote_path=None):
762
742
"""Creates a client that will connect on the first use.
764
:param ssh_params: A SSHParams instance.
765
744
:param vendor: An optional override for the ssh vendor to use. See
766
745
bzrlib.transport.ssh for details on ssh vendors.
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'
747
self._connected = False
749
self._password = password
751
self._username = username
773
752
# SmartClientStreamMedium stores the repr of this object in its
774
753
# _DebugCounter so we have to store all the values used in our repr
775
754
# method before calling the super init.
776
755
SmartClientStreamMedium.__init__(self, base)
756
self._read_from = None
757
self._ssh_connection = None
777
758
self._vendor = vendor
778
self._ssh_connection = None
759
self._write_to = None
760
self._bzr_remote_path = bzr_remote_path
761
# for the benefit of progress making a short description of this
763
self._scheme = 'bzr+ssh'
780
765
def __repr__(self):
781
if self._ssh_params.port is None:
784
maybe_port = ':%s' % self._ssh_params.port
785
return "%s(%s://%s@%s%s/)" % (
766
return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
786
767
self.__class__.__name__,
788
self._ssh_params.username,
789
self._ssh_params.host,
792
773
def _accept_bytes(self, bytes):
793
774
"""See SmartClientStreamMedium.accept_bytes."""
794
775
self._ensure_connection()
795
self._real_medium.accept_bytes(bytes)
776
osutils.until_no_eintr(self._write_to.write, bytes)
777
self._report_activity(len(bytes), 'write')
797
779
def disconnect(self):
798
780
"""See SmartClientMedium.disconnect()."""
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
781
if not self._connected:
783
osutils.until_no_eintr(self._read_from.close)
784
osutils.until_no_eintr(self._write_to.close)
785
self._ssh_connection.close()
786
self._connected = False
806
788
def _ensure_connection(self):
807
789
"""Connect this medium if not already connected."""
808
if self._real_medium is not None:
810
792
if self._vendor is None:
811
793
vendor = ssh._get_ssh_vendor()
813
795
vendor = self._vendor
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',
796
self._ssh_connection = vendor.connect_ssh(self._username,
797
self._password, self._host, self._port,
798
command=[self._bzr_remote_path, 'serve', '--inet',
818
799
'--directory=/', '--allow-writes'])
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))
800
self._read_from, self._write_to = \
801
self._ssh_connection.get_filelike_channels()
802
self._connected = True
832
804
def _flush(self):
833
805
"""See SmartClientStreamMedium._flush()."""
834
self._real_medium._flush()
806
self._write_to.flush()
836
808
def _read_bytes(self, count):
837
809
"""See SmartClientStreamMedium.read_bytes."""
838
if self._real_medium is None:
810
if not self._connected:
839
811
raise errors.MediumNotConnected(self)
840
return self._real_medium.read_bytes(count)
812
bytes_to_read = min(count, _MAX_READ_SIZE)
813
bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
814
self._report_activity(len(bytes), 'read')
843
818
# Port 4155 is the default port for bzr://, registered with IANA.
845
820
BZR_DEFAULT_PORT = 4155
848
class SmartClientSocketMedium(SmartClientStreamMedium):
849
"""A client medium using a socket.
851
This class isn't usable directly. Use one of its subclasses instead.
823
class SmartTCPClientMedium(SmartClientStreamMedium):
824
"""A client medium using TCP."""
854
def __init__(self, base):
826
def __init__(self, host, port, base):
827
"""Creates a client that will connect on the first use."""
855
828
SmartClientStreamMedium.__init__(self, base)
829
self._connected = False
856
832
self._socket = None
857
self._connected = False
859
834
def _accept_bytes(self, bytes):
860
835
"""See SmartClientMedium.accept_bytes."""
861
836
self._ensure_connection()
862
837
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)
883
839
def disconnect(self):
884
840
"""See SmartClientMedium.disconnect()."""
885
841
if not self._connected:
843
osutils.until_no_eintr(self._socket.close)
888
844
self._socket = None
889
845
self._connected = False
892
class SmartTCPClientMedium(SmartClientSocketMedium):
893
"""A client medium that creates a TCP connection."""
895
def __init__(self, host, port, base):
896
"""Creates a client that will connect on the first use."""
897
SmartClientSocketMedium.__init__(self, base)
901
847
def _ensure_connection(self):
902
848
"""Connect this medium if not already connected."""
903
849
if self._connected:
937
883
(self._host, port, err_msg))
938
884
self._connected = True
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.
887
"""See SmartClientStreamMedium._flush().
889
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
890
add a means to do a flush, but that can be done in the future.
893
def _read_bytes(self, count):
894
"""See SmartClientMedium.read_bytes."""
895
if not self._connected:
896
raise errors.MediumNotConnected(self)
897
return _read_bytes_from_socket(
898
self._socket.recv, count, self._report_activity)
958
901
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
added
removed
bzrlib/smart/medium.py
