284
285
self._push_back(protocol.unused_data)
286
287
def _read_bytes(self, desired_count):
287
return osutils.read_bytes_from_socket(
288
self.socket, self._report_activity)
288
return _read_bytes_from_socket(
289
self.socket.recv, desired_count, self._report_activity)
290
291
def terminate_due_to_error(self):
291
292
# TODO: This should log to a server log file, but no such thing
292
293
# exists yet. Andrew Bennetts 2006-09-29.
294
osutils.until_no_eintr(self.socket.close)
294
295
self.finished = True
296
297
def _write_out(self, bytes):
297
tstart = osutils.timer_func()
298
298
osutils.send_all(self.socket, bytes, self._report_activity)
299
if 'hpss' in debug.debug_flags:
300
thread_id = thread.get_ident()
301
trace.mutter('%12s: [%s] %d bytes to the socket in %.3fs'
302
% ('wrote', thread_id, len(bytes),
303
osutils.timer_func() - tstart))
306
301
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
724
713
def _accept_bytes(self, bytes):
725
714
"""See SmartClientStreamMedium.accept_bytes."""
726
self._writeable_pipe.write(bytes)
715
osutils.until_no_eintr(self._writeable_pipe.write, bytes)
727
716
self._report_activity(len(bytes), 'write')
729
718
def _flush(self):
730
719
"""See SmartClientStreamMedium._flush()."""
731
self._writeable_pipe.flush()
720
osutils.until_no_eintr(self._writeable_pipe.flush)
733
722
def _read_bytes(self, count):
734
723
"""See SmartClientStreamMedium._read_bytes."""
735
bytes_to_read = min(count, _MAX_READ_SIZE)
736
bytes = self._readable_pipe.read(bytes_to_read)
724
bytes = osutils.until_no_eintr(self._readable_pipe.read, count)
737
725
self._report_activity(len(bytes), 'read')
741
class SSHParams(object):
742
"""A set of parameters for starting a remote bzr via SSH."""
729
class SmartSSHClientMedium(SmartClientStreamMedium):
730
"""A client medium using SSH."""
744
732
def __init__(self, host, port=None, username=None, password=None,
745
bzr_remote_path='bzr'):
748
self.username = username
749
self.password = password
750
self.bzr_remote_path = bzr_remote_path
753
class SmartSSHClientMedium(SmartClientStreamMedium):
754
"""A client medium using SSH.
756
It delegates IO to a SmartClientSocketMedium or
757
SmartClientAlreadyConnectedSocketMedium (depending on platform).
760
def __init__(self, base, ssh_params, vendor=None):
733
base=None, vendor=None, bzr_remote_path=None):
761
734
"""Creates a client that will connect on the first use.
763
:param ssh_params: A SSHParams instance.
764
736
:param vendor: An optional override for the ssh vendor to use. See
765
737
bzrlib.transport.ssh for details on ssh vendors.
767
self._real_medium = None
768
self._ssh_params = ssh_params
769
# for the benefit of progress making a short description of this
771
self._scheme = 'bzr+ssh'
739
self._connected = False
741
self._password = password
743
self._username = username
772
744
# SmartClientStreamMedium stores the repr of this object in its
773
745
# _DebugCounter so we have to store all the values used in our repr
774
746
# method before calling the super init.
775
747
SmartClientStreamMedium.__init__(self, base)
748
self._read_from = None
749
self._ssh_connection = None
776
750
self._vendor = vendor
777
self._ssh_connection = None
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'
779
757
def __repr__(self):
780
if self._ssh_params.port is None:
783
maybe_port = ':%s' % self._ssh_params.port
784
return "%s(%s://%s@%s%s/)" % (
758
return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
785
759
self.__class__.__name__,
787
self._ssh_params.username,
788
self._ssh_params.host,
791
765
def _accept_bytes(self, bytes):
792
766
"""See SmartClientStreamMedium.accept_bytes."""
793
767
self._ensure_connection()
794
self._real_medium.accept_bytes(bytes)
768
osutils.until_no_eintr(self._write_to.write, bytes)
769
self._report_activity(len(bytes), 'write')
796
771
def disconnect(self):
797
772
"""See SmartClientMedium.disconnect()."""
798
if self._real_medium is not None:
799
self._real_medium.disconnect()
800
self._real_medium = None
801
if self._ssh_connection is not None:
802
self._ssh_connection.close()
803
self._ssh_connection = None
773
if not self._connected:
775
osutils.until_no_eintr(self._read_from.close)
776
osutils.until_no_eintr(self._write_to.close)
777
self._ssh_connection.close()
778
self._connected = False
805
780
def _ensure_connection(self):
806
781
"""Connect this medium if not already connected."""
807
if self._real_medium is not None:
809
784
if self._vendor is None:
810
785
vendor = ssh._get_ssh_vendor()
812
787
vendor = self._vendor
813
self._ssh_connection = vendor.connect_ssh(self._ssh_params.username,
814
self._ssh_params.password, self._ssh_params.host,
815
self._ssh_params.port,
816
command=[self._ssh_params.bzr_remote_path, 'serve', '--inet',
788
self._ssh_connection = vendor.connect_ssh(self._username,
789
self._password, self._host, self._port,
790
command=[self._bzr_remote_path, 'serve', '--inet',
817
791
'--directory=/', '--allow-writes'])
818
io_kind, io_object = self._ssh_connection.get_sock_or_pipes()
819
if io_kind == 'socket':
820
self._real_medium = SmartClientAlreadyConnectedSocketMedium(
821
self.base, io_object)
822
elif io_kind == 'pipes':
823
read_from, write_to = io_object
824
self._real_medium = SmartSimplePipesClientMedium(
825
read_from, write_to, self.base)
827
raise AssertionError(
828
"Unexpected io_kind %r from %r"
829
% (io_kind, self._ssh_connection))
792
self._read_from, self._write_to = \
793
self._ssh_connection.get_filelike_channels()
794
self._connected = True
831
796
def _flush(self):
832
797
"""See SmartClientStreamMedium._flush()."""
833
self._real_medium._flush()
798
self._write_to.flush()
835
800
def _read_bytes(self, count):
836
801
"""See SmartClientStreamMedium.read_bytes."""
837
if self._real_medium is None:
802
if not self._connected:
838
803
raise errors.MediumNotConnected(self)
839
return self._real_medium.read_bytes(count)
804
bytes_to_read = min(count, _MAX_READ_SIZE)
805
bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
806
self._report_activity(len(bytes), 'read')
842
810
# Port 4155 is the default port for bzr://, registered with IANA.
844
812
BZR_DEFAULT_PORT = 4155
847
class SmartClientSocketMedium(SmartClientStreamMedium):
848
"""A client medium using a socket.
850
This class isn't usable directly. Use one of its subclasses instead.
815
class SmartTCPClientMedium(SmartClientStreamMedium):
816
"""A client medium using TCP."""
853
def __init__(self, base):
818
def __init__(self, host, port, base):
819
"""Creates a client that will connect on the first use."""
854
820
SmartClientStreamMedium.__init__(self, base)
821
self._connected = False
855
824
self._socket = None
856
self._connected = False
858
826
def _accept_bytes(self, bytes):
859
827
"""See SmartClientMedium.accept_bytes."""
860
828
self._ensure_connection()
861
829
osutils.send_all(self._socket, bytes, self._report_activity)
863
def _ensure_connection(self):
864
"""Connect this medium if not already connected."""
865
raise NotImplementedError(self._ensure_connection)
868
"""See SmartClientStreamMedium._flush().
870
For sockets we do no flushing. For TCP sockets we may want to turn off
871
TCP_NODELAY and add a means to do a flush, but that can be done in the
875
def _read_bytes(self, count):
876
"""See SmartClientMedium.read_bytes."""
877
if not self._connected:
878
raise errors.MediumNotConnected(self)
879
return osutils.read_bytes_from_socket(
880
self._socket, self._report_activity)
882
831
def disconnect(self):
883
832
"""See SmartClientMedium.disconnect()."""
884
833
if not self._connected:
835
osutils.until_no_eintr(self._socket.close)
887
836
self._socket = None
888
837
self._connected = False
891
class SmartTCPClientMedium(SmartClientSocketMedium):
892
"""A client medium that creates a TCP connection."""
894
def __init__(self, host, port, base):
895
"""Creates a client that will connect on the first use."""
896
SmartClientSocketMedium.__init__(self, base)
900
839
def _ensure_connection(self):
901
840
"""Connect this medium if not already connected."""
902
841
if self._connected:
936
875
(self._host, port, err_msg))
937
876
self._connected = True
940
class SmartClientAlreadyConnectedSocketMedium(SmartClientSocketMedium):
941
"""A client medium for an already connected socket.
943
Note that this class will assume it "owns" the socket, so it will close it
944
when its disconnect method is called.
947
def __init__(self, base, sock):
948
SmartClientSocketMedium.__init__(self, base)
950
self._connected = True
952
def _ensure_connection(self):
953
# Already connected, by definition! So nothing to do.
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)
957
893
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
added
removed
bzrlib/smart/medium.py
