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