491
491
return self._medium._get_line()
494
class _VfsRefuser(object):
495
"""An object that refuses all VFS requests.
500
client._SmartClient.hooks.install_named_hook(
501
'call', self.check_vfs, 'vfs refuser')
503
def check_vfs(self, params):
505
request_method = request.request_handlers.get(params.method)
507
# A method we don't know about doesn't count as a VFS method.
509
if issubclass(request_method, vfs.VfsRequest):
510
raise errors.HpssVfsRequestNotAllowed(params.method, params.args)
513
494
class _DebugCounter(object):
514
495
"""An object that counts the HPSS calls made to each client medium.
516
When a medium is garbage-collected, or failing that when
517
bzrlib.global_state exits, the total number of calls made on that medium
518
are reported via trace.note.
497
When a medium is garbage-collected, or failing that when atexit functions
498
are run, the total number of calls made on that medium are reported via
521
502
def __init__(self):
522
503
self.counts = weakref.WeakKeyDictionary()
523
504
client._SmartClient.hooks.install_named_hook(
524
505
'call', self.increment_call_count, 'hpss call counter')
525
bzrlib.global_state.cleanups.add_cleanup(self.flush_all)
506
atexit.register(self.flush_all)
527
508
def track(self, medium):
528
509
"""Start tracking calls made to a medium.
758
738
def _read_bytes(self, count):
759
739
"""See SmartClientStreamMedium._read_bytes."""
760
bytes_to_read = min(count, _MAX_READ_SIZE)
761
bytes = self._readable_pipe.read(bytes_to_read)
740
bytes = osutils.until_no_eintr(self._readable_pipe.read, count)
762
741
self._report_activity(len(bytes), 'read')
766
class SSHParams(object):
767
"""A set of parameters for starting a remote bzr via SSH."""
745
class SmartSSHClientMedium(SmartClientStreamMedium):
746
"""A client medium using SSH."""
769
748
def __init__(self, host, port=None, username=None, password=None,
770
bzr_remote_path='bzr'):
773
self.username = username
774
self.password = password
775
self.bzr_remote_path = bzr_remote_path
778
class SmartSSHClientMedium(SmartClientStreamMedium):
779
"""A client medium using SSH.
781
It delegates IO to a SmartClientSocketMedium or
782
SmartClientAlreadyConnectedSocketMedium (depending on platform).
785
def __init__(self, base, ssh_params, vendor=None):
749
base=None, vendor=None, bzr_remote_path=None):
786
750
"""Creates a client that will connect on the first use.
788
:param ssh_params: A SSHParams instance.
789
752
:param vendor: An optional override for the ssh vendor to use. See
790
753
bzrlib.transport.ssh for details on ssh vendors.
792
self._real_medium = None
793
self._ssh_params = ssh_params
755
self._connected = False
757
self._password = password
759
self._username = username
794
760
# for the benefit of progress making a short description of this
796
762
self._scheme = 'bzr+ssh'
798
764
# _DebugCounter so we have to store all the values used in our repr
799
765
# method before calling the super init.
800
766
SmartClientStreamMedium.__init__(self, base)
767
self._read_from = None
768
self._ssh_connection = None
801
769
self._vendor = vendor
802
self._ssh_connection = None
770
self._write_to = None
771
self._bzr_remote_path = bzr_remote_path
804
773
def __repr__(self):
805
if self._ssh_params.port is None:
774
if self._port is None:
808
maybe_port = ':%s' % self._ssh_params.port
777
maybe_port = ':%s' % self._port
809
778
return "%s(%s://%s@%s%s/)" % (
810
779
self.__class__.__name__,
812
self._ssh_params.username,
813
self._ssh_params.host,
816
785
def _accept_bytes(self, bytes):
817
786
"""See SmartClientStreamMedium.accept_bytes."""
818
787
self._ensure_connection()
819
self._real_medium.accept_bytes(bytes)
788
self._write_to.write(bytes)
789
self._report_activity(len(bytes), 'write')
821
791
def disconnect(self):
822
792
"""See SmartClientMedium.disconnect()."""
823
if self._real_medium is not None:
824
self._real_medium.disconnect()
825
self._real_medium = None
826
if self._ssh_connection is not None:
827
self._ssh_connection.close()
828
self._ssh_connection = None
793
if not self._connected:
795
self._read_from.close()
796
self._write_to.close()
797
self._ssh_connection.close()
798
self._connected = False
830
800
def _ensure_connection(self):
831
801
"""Connect this medium if not already connected."""
832
if self._real_medium is not None:
834
804
if self._vendor is None:
835
805
vendor = ssh._get_ssh_vendor()
837
807
vendor = self._vendor
838
self._ssh_connection = vendor.connect_ssh(self._ssh_params.username,
839
self._ssh_params.password, self._ssh_params.host,
840
self._ssh_params.port,
841
command=[self._ssh_params.bzr_remote_path, 'serve', '--inet',
808
self._ssh_connection = vendor.connect_ssh(self._username,
809
self._password, self._host, self._port,
810
command=[self._bzr_remote_path, 'serve', '--inet',
842
811
'--directory=/', '--allow-writes'])
843
io_kind, io_object = self._ssh_connection.get_sock_or_pipes()
844
if io_kind == 'socket':
845
self._real_medium = SmartClientAlreadyConnectedSocketMedium(
846
self.base, io_object)
847
elif io_kind == 'pipes':
848
read_from, write_to = io_object
849
self._real_medium = SmartSimplePipesClientMedium(
850
read_from, write_to, self.base)
852
raise AssertionError(
853
"Unexpected io_kind %r from %r"
854
% (io_kind, self._ssh_connection))
812
self._read_from, self._write_to = \
813
self._ssh_connection.get_filelike_channels()
814
self._connected = True
856
816
def _flush(self):
857
817
"""See SmartClientStreamMedium._flush()."""
858
self._real_medium._flush()
818
self._write_to.flush()
860
820
def _read_bytes(self, count):
861
821
"""See SmartClientStreamMedium.read_bytes."""
862
if self._real_medium is None:
822
if not self._connected:
863
823
raise errors.MediumNotConnected(self)
864
return self._real_medium.read_bytes(count)
824
bytes_to_read = min(count, _MAX_READ_SIZE)
825
bytes = self._read_from.read(bytes_to_read)
826
self._report_activity(len(bytes), 'read')
867
830
# Port 4155 is the default port for bzr://, registered with IANA.
869
832
BZR_DEFAULT_PORT = 4155
872
class SmartClientSocketMedium(SmartClientStreamMedium):
873
"""A client medium using a socket.
875
This class isn't usable directly. Use one of its subclasses instead.
835
class SmartTCPClientMedium(SmartClientStreamMedium):
836
"""A client medium using TCP."""
878
def __init__(self, base):
838
def __init__(self, host, port, base):
839
"""Creates a client that will connect on the first use."""
879
840
SmartClientStreamMedium.__init__(self, base)
841
self._connected = False
880
844
self._socket = None
881
self._connected = False
883
846
def _accept_bytes(self, bytes):
884
847
"""See SmartClientMedium.accept_bytes."""
885
848
self._ensure_connection()
886
849
osutils.send_all(self._socket, bytes, self._report_activity)
888
def _ensure_connection(self):
889
"""Connect this medium if not already connected."""
890
raise NotImplementedError(self._ensure_connection)
893
"""See SmartClientStreamMedium._flush().
895
For sockets we do no flushing. For TCP sockets we may want to turn off
896
TCP_NODELAY and add a means to do a flush, but that can be done in the
900
def _read_bytes(self, count):
901
"""See SmartClientMedium.read_bytes."""
902
if not self._connected:
903
raise errors.MediumNotConnected(self)
904
return osutils.read_bytes_from_socket(
905
self._socket, self._report_activity)
907
851
def disconnect(self):
908
852
"""See SmartClientMedium.disconnect()."""
909
853
if not self._connected:
961
895
(self._host, port, err_msg))
962
896
self._connected = True
965
class SmartClientAlreadyConnectedSocketMedium(SmartClientSocketMedium):
966
"""A client medium for an already connected socket.
968
Note that this class will assume it "owns" the socket, so it will close it
969
when its disconnect method is called.
972
def __init__(self, base, sock):
973
SmartClientSocketMedium.__init__(self, base)
975
self._connected = True
977
def _ensure_connection(self):
978
# Already connected, by definition! So nothing to do.
899
"""See SmartClientStreamMedium._flush().
901
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
902
add a means to do a flush, but that can be done in the future.
905
def _read_bytes(self, count):
906
"""See SmartClientMedium.read_bytes."""
907
if not self._connected:
908
raise errors.MediumNotConnected(self)
909
return osutils.read_bytes_from_socket(
910
self._socket, self._report_activity)
982
913
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
added
removed
bzrlib/smart/medium.py
