<s> from __future__ import absolute_import , division , print_function , unicode_literals <EOL> __metaclass__ = type <EOL> import errno <EOL> import six <EOL> import socket <EOL> import ssl <EOL> from abc import ABCMeta , abstractmethod <EOL> import logging <EOL> from threading import RLock <EOL> from ssl import SSLError <EOL> import datetime <EOL> import time <EOL> from . import compat <EOL> from . proto import Frame <EOL> from . concurrency import synchronized <EOL> from . exceptions import UnexpectedFrame <EOL> from . utils import get_errno <EOL> from . spec import FrameType <EOL> log = logging . getLogger ( '<STR_LIT>' ) <EOL> compat . patch ( ) <EOL> _UNAVAIL = { errno . EAGAIN , errno . EINTR , errno . ENOENT } <EOL> AMQP_PROTOCOL_HEADER = b'<STR_LIT>' <EOL> class Transport : <EOL> __metaclass__ = ABCMeta <EOL> """<STR_LIT>""" <EOL> connected = False <EOL> def __init__ ( self , host , port , connect_timeout , buf_size ) : <EOL> """<STR_LIT>""" <EOL> self . _rbuf = bytearray ( buf_size ) <EOL> self . last_heartbeat_sent = None <EOL> self . last_heartbeat_received = None <EOL> self . last_heartbeat_sent_monotonic = <NUM_LIT:0.0> <EOL> self . _frame_write_lock = RLock ( ) <EOL> self . _frame_read_lock = RLock ( ) <EOL> self . sock = None <EOL> last_err = None <EOL> for res in socket . getaddrinfo ( host , port , <NUM_LIT:0> , socket . SOCK_STREAM , socket . IPPROTO_TCP ) : <EOL> af , socktype , proto , canonname , sa = res <EOL> try : <EOL> self . sock = socket . socket ( af , socktype , proto ) <EOL> self . sock . settimeout ( connect_timeout ) <EOL> self . sock . connect ( sa ) <EOL> break <EOL> except socket . error as exc : <EOL> self . sock . close ( ) <EOL> self . sock = None <EOL> last_err = exc <EOL> if not self . sock : <EOL> raise socket . error ( last_err ) <EOL> try : <EOL> assert isinstance ( self . sock , socket . socket ) <EOL> self . sock . settimeout ( None ) <EOL> self . sock . setsockopt ( socket . IPPROTO_TCP , socket . TCP_NODELAY , <NUM_LIT:1> ) <EOL> self . sock . setsockopt ( socket . SOL_SOCKET , socket . SO_KEEPALIVE , <NUM_LIT:1> ) <EOL> self . _setup_transport ( ) <EOL> self . write ( AMQP_PROTOCOL_HEADER ) <EOL> except ( OSError , IOError , socket . error ) as exc : <EOL> if get_errno ( exc ) not in _UNAVAIL : <EOL> self . connected = False <EOL> raise <EOL> self . connected = True <EOL> def __del__ ( self ) : <EOL> try : <EOL> if socket is not None : <EOL> try : <EOL> self . close ( ) <EOL> except : <EOL> pass <EOL> finally : <EOL> self . sock = None <EOL> def _read ( self , n , initial , _errnos ) : <EOL> """<STR_LIT>""" <EOL> mview = memoryview ( self . _rbuf ) <EOL> to_read = n <EOL> while to_read : <EOL> try : <EOL> bytes_read = self . sock . recv_into ( mview , to_read ) <EOL> mview = mview [ bytes_read : ] <EOL> to_read -= bytes_read <EOL> except socket . error as exc : <EOL> if not initial and exc . errno in _errnos : <EOL> continue <EOL> raise <EOL> if not bytes_read : <EOL> raise IOError ( '<STR_LIT>' ) <EOL> return memoryview ( self . _rbuf ) [ : n ] <EOL> @ abstractmethod <EOL> def read ( self , n , initial = False ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ abstractmethod <EOL> def write ( self , s ) : <EOL> """<STR_LIT>""" <EOL> def _setup_transport ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def close ( self ) : <EOL> if self . sock is not None : <EOL> self . sock . shutdown ( socket . SHUT_RDWR ) <EOL> self . sock . close ( ) <EOL> self . sock = None <EOL> self . connected = False <EOL> @ synchronized ( '<STR_LIT>' ) <EOL> def read_frame ( self ) : <EOL> """<STR_LIT>""" <EOL> frame = Frame ( ) <EOL> try : <EOL> frame_header = self . read ( <NUM_LIT:7> , True ) <EOL> frame . data . extend ( frame_header ) <EOL> payload = self . read ( frame . payload_size ) <EOL> frame . data . extend ( payload ) <EOL> frame_terminator = self . read ( <NUM_LIT:1> ) <EOL> frame . data . extend ( frame_terminator ) <EOL> if six . PY2 : <EOL> i_last_byte = six . byte2int ( frame_terminator ) <EOL> else : <EOL> i_last_byte = six . byte2int ( bytes ( frame_terminator ) ) <EOL> except ( OSError , IOError , socket . error ) as exc : <EOL> if isinstance ( exc , SSLError ) and '<STR_LIT>' in str ( exc ) : <EOL> raise socket . timeout ( ) <EOL> if get_errno ( exc ) not in _UNAVAIL and not isinstance ( exc , socket . timeout ) : <EOL> self . connected = False <EOL> raise <EOL> if i_last_byte == FrameType . END : <EOL> if frame . frame_type == FrameType . HEARTBEAT : <EOL> self . last_heartbeat_received = datetime . datetime . now ( ) <EOL> return frame <EOL> else : <EOL> raise UnexpectedFrame ( '<STR_LIT>' . format ( hex ( i_last_byte ) ) ) <EOL> @ synchronized ( '<STR_LIT>' ) <EOL> def write_frame ( self , frame ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . write ( frame . data ) <EOL> except socket . timeout : <EOL> raise <EOL> except ( OSError , IOError , socket . error ) as exc : <EOL> if get_errno ( exc ) not in _UNAVAIL : <EOL> self . connected = False <EOL> raise <EOL> def send_heartbeat ( self ) : <EOL> """<STR_LIT>""" <EOL> self . last_heartbeat_sent = datetime . datetime . now ( ) <EOL> self . last_heartbeat_sent_monotonic = time . monotonic ( ) <EOL> self . write_frame ( Frame ( FrameType . HEARTBEAT ) ) <EOL> def is_alive ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . sock : <EOL> return False <EOL> if not self . connected : <EOL> return False <EOL> try : <EOL> self . send_heartbeat ( ) <EOL> except socket . error : <EOL> return False <EOL> return True <EOL> class SSLTransport ( Transport ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , host , port , connect_timeout , frame_max , ssl_opts ) : <EOL> self . ssl_opts = ssl_opts <EOL> super ( SSLTransport , self ) . __init__ ( host , port , connect_timeout , frame_max ) <EOL> def _setup_transport ( self ) : <EOL> """<STR_LIT>""" <EOL> self . sock = ssl . wrap_socket ( self . sock , ** self . ssl_opts ) <EOL> def read ( self , n , initial = False ) : <EOL> """<STR_LIT>""" <EOL> return self . _read ( n , initial , _errnos = ( errno . ENOENT , errno . EAGAIN , errno . EINTR ) ) <EOL> def write ( self , s ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> write = self . sock . write <EOL> except AttributeError : <EOL> raise IOError ( '<STR_LIT>' ) <EOL> else : <EOL> while s : <EOL> n = write ( s ) <EOL> if not n : <EOL> raise IOError ( '<STR_LIT>' ) <EOL> s = s [ n : ] <EOL> class TCPTransport ( Transport ) : <EOL> """<STR_LIT>""" <EOL> def read ( self , n , initial = False ) : <EOL> """<STR_LIT>""" <EOL> return self . _read ( n , initial , _errnos = ( errno . EAGAIN , errno . EINTR ) ) <EOL> def write ( self , s ) : <EOL> self . sock . sendall ( s ) <EOL> def create_transport ( host , port , connect_timeout , frame_max , ssl_opts = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( ssl_opts , dict ) : <EOL> return SSLTransport ( host , port , connect_timeout , frame_max , ssl_opts ) <EOL> else : <EOL> return TCPTransport ( host , port , connect_timeout , frame_max ) </s>
