<s> """<STR_LIT>""" <EOL> import errno <EOL> import functools <EOL> import os <EOL> import re <EOL> import socket <EOL> import ssl <EOL> import struct <EOL> from pants . _channel import _Channel , HAS_IPV6 , sock_type <EOL> from pants . engine import Engine <EOL> try : <EOL> from netstruct import NetStruct as _NetStruct <EOL> except ImportError : <EOL> class _NetStruct ( object ) : <EOL> def __init__ ( self , * a , ** kw ) : <EOL> raise NotImplementedError <EOL> RegexType = type ( re . compile ( "<STR_LIT>" ) ) <EOL> Struct = struct . Struct <EOL> import logging <EOL> log = logging . getLogger ( "<STR_LIT>" ) <EOL> class Stream ( _Channel ) : <EOL> """<STR_LIT>""" <EOL> SEND_STRING = <NUM_LIT:0> <EOL> SEND_FILE = <NUM_LIT:1> <EOL> SEND_SSL_HANDSHAKE = <NUM_LIT:2> <EOL> def __init__ ( self , ** kwargs ) : <EOL> sock = kwargs . get ( "<STR_LIT>" , None ) <EOL> if sock and sock_type ( sock ) != socket . SOCK_STREAM : <EOL> raise TypeError ( "<STR_LIT>" <EOL> % self . __class__ . __name__ ) <EOL> _Channel . __init__ ( self , ** kwargs ) <EOL> self . _remote_address = None <EOL> self . _local_address = None <EOL> self . _read_delimiter = None <EOL> self . _recv_buffer = "<STR_LIT>" <EOL> self . _recv_buffer_size_limit = self . _buffer_size <EOL> self . _send_buffer = [ ] <EOL> self . connected = False <EOL> self . connecting = False <EOL> self . _closing = False <EOL> self . ssl_enabled = False <EOL> self . _ssl_enabling = False <EOL> self . _ssl_socket_wrapped = False <EOL> self . _ssl_handshake_done = False <EOL> self . _ssl_call_on_connect = False <EOL> if isinstance ( kwargs . get ( "<STR_LIT>" , None ) , ssl . SSLSocket ) : <EOL> self . _ssl_socket_wrapped = True <EOL> self . startSSL ( ) <EOL> elif kwargs . get ( "<STR_LIT>" , None ) is not None : <EOL> self . startSSL ( kwargs [ "<STR_LIT>" ] ) <EOL> @ property <EOL> def remote_address ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _remote_address is not None : <EOL> return self . _remote_address <EOL> elif self . _socket : <EOL> try : <EOL> return self . _socket . getpeername ( ) <EOL> except socket . error : <EOL> return None <EOL> else : <EOL> return None <EOL> @ remote_address . setter <EOL> def remote_address ( self , val ) : <EOL> self . _remote_address = val <EOL> @ remote_address . deleter <EOL> def remote_address ( self ) : <EOL> self . _remote_address = None <EOL> @ property <EOL> def local_address ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _local_address is not None : <EOL> return self . _local_address <EOL> elif self . _socket : <EOL> try : <EOL> return self . _socket . getsockname ( ) <EOL> except socket . error : <EOL> return None <EOL> else : <EOL> return None <EOL> @ local_address . setter <EOL> def local_address ( self , val ) : <EOL> self . _local_address = val <EOL> @ local_address . deleter <EOL> def local_address ( self ) : <EOL> self . _local_address = None <EOL> @ property <EOL> def read_delimiter ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _read_delimiter <EOL> @ read_delimiter . setter <EOL> def read_delimiter ( self , value ) : <EOL> if value is None or isinstance ( value , basestring ) or isinstance ( value , RegexType ) : <EOL> self . _read_delimiter = value <EOL> self . _recv_buffer_size_limit = self . _buffer_size <EOL> elif isinstance ( value , ( int , long ) ) : <EOL> self . _read_delimiter = value <EOL> self . _recv_buffer_size_limit = max ( self . _buffer_size , value ) <EOL> elif isinstance ( value , Struct ) : <EOL> self . _read_delimiter = value <EOL> self . _recv_buffer_size_limit = max ( self . _buffer_size , value . size ) <EOL> elif isinstance ( value , _NetStruct ) : <EOL> self . _read_delimiter = value <EOL> self . _recv_buffer_size_limit = max ( self . _buffer_size , <EOL> value . minimum_size ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . _netstruct_iter = None <EOL> self . _netstruct_needed = None <EOL> regex_search = True <EOL> _buffer_size = <NUM_LIT:2> ** <NUM_LIT:16> <EOL> @ property <EOL> def buffer_size ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _buffer_size <EOL> @ buffer_size . setter <EOL> def buffer_size ( self , value ) : <EOL> if not isinstance ( value , ( long , int ) ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . _buffer_size = value <EOL> if isinstance ( self . _read_delimiter , ( int , long ) ) : <EOL> self . _recv_buffer_size_limit = max ( value , self . _read_delimiter ) <EOL> elif isinstance ( self . _read_delimiter , Struct ) : <EOL> self . _recv_buffer_size_limit = max ( value , <EOL> self . _read_delimiter . size ) <EOL> elif isinstance ( self . _read_delimiter , _NetStruct ) : <EOL> self . _recv_buffer_size_limit = max ( value , <EOL> self . _read_delimiter . minimum_size ) <EOL> else : <EOL> self . _recv_buffer_size_limit = value <EOL> def startSSL ( self , ssl_options = { } ) : <EOL> """<STR_LIT>""" <EOL> if self . ssl_enabled or self . _ssl_enabling : <EOL> raise RuntimeError ( "<STR_LIT>" % self ) <EOL> if self . _closed or self . _closing : <EOL> raise RuntimeError ( "<STR_LIT>" % self ) <EOL> if ssl_options . setdefault ( "<STR_LIT>" , False ) is not False : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . _ssl_enabling = True <EOL> self . _send_buffer . append ( ( Stream . SEND_SSL_HANDSHAKE , ssl_options ) ) <EOL> if self . connected : <EOL> self . _process_send_buffer ( ) <EOL> return self <EOL> def connect ( self , address ) : <EOL> """<STR_LIT>""" <EOL> if self . connected or self . connecting : <EOL> raise RuntimeError ( "<STR_LIT>" % self ) <EOL> if self . _closed or self . _closing : <EOL> raise RuntimeError ( "<STR_LIT>" % self ) <EOL> self . connecting = True <EOL> address , family , resolved = self . _format_address ( address ) <EOL> if resolved : <EOL> self . _do_connect ( address , family ) <EOL> else : <EOL> try : <EOL> result = socket . getaddrinfo ( address [ <NUM_LIT:0> ] , address [ <NUM_LIT:1> ] , family ) <EOL> except socket . error as err : <EOL> self . close ( flush = False ) <EOL> e = StreamConnectError ( err . errno , err . strerror ) <EOL> self . _safely_call ( self . on_connect_error , e ) <EOL> return self <EOL> result = result [ <NUM_LIT:0> ] <EOL> self . _do_connect ( result [ - <NUM_LIT:1> ] , result [ <NUM_LIT:0> ] ) <EOL> return self <EOL> def close ( self , flush = True ) : <EOL> """<STR_LIT>""" <EOL> if self . _closed : <EOL> return <EOL> if flush and self . _send_buffer : <EOL> self . _closing = True <EOL> return <EOL> self . read_delimiter = None <EOL> self . _recv_buffer = "<STR_LIT>" <EOL> self . _send_buffer = [ ] <EOL> self . connected = False <EOL> self . connecting = False <EOL> self . ssl_enabled = False <EOL> self . _ssl_enabling = False <EOL> self . _ssl_socket_wrapped = False <EOL> self . _ssl_handshake_done = False <EOL> self . _ssl_call_on_connect = False <EOL> self . _safely_call ( self . on_close ) <EOL> self . _remote_address = None <EOL> self . _local_address = None <EOL> _Channel . close ( self ) <EOL> self . _closing = False <EOL> def write ( self , data , flush = False ) : <EOL> """<STR_LIT>""" <EOL> if self . _closed or self . _closing : <EOL> raise RuntimeError ( "<STR_LIT>" % self ) <EOL> if not self . connected : <EOL> raise RuntimeError ( "<STR_LIT>" % self ) <EOL> if self . _send_buffer and self . _send_buffer [ - <NUM_LIT:1> ] [ <NUM_LIT:0> ] == Stream . SEND_STRING : <EOL> data_type , existing_data = self . _send_buffer . pop ( - <NUM_LIT:1> ) <EOL> data = existing_data + data <EOL> self . _send_buffer . append ( ( Stream . SEND_STRING , data ) ) <EOL> if flush : <EOL> self . _process_send_buffer ( ) <EOL> else : <EOL> self . _start_waiting_for_write_event ( ) <EOL> def write_file ( self , sfile , nbytes = <NUM_LIT:0> , offset = <NUM_LIT:0> , flush = False ) : <EOL> """<STR_LIT>""" <EOL> if self . _closed or self . _closing : <EOL> raise RuntimeError ( "<STR_LIT>" % self ) <EOL> if not self . connected : <EOL> raise RuntimeError ( "<STR_LIT>" % self ) <EOL> self . _send_buffer . append ( ( Stream . SEND_FILE , ( sfile , offset , nbytes ) ) ) <EOL> if flush : <EOL> self . _process_send_buffer ( ) <EOL> else : <EOL> self . _start_waiting_for_write_event ( ) <EOL> def write_packed ( self , * data , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> format = kwargs . get ( "<STR_LIT>" ) <EOL> if format : <EOL> self . write ( struct . pack ( format , * data ) , kwargs . get ( "<STR_LIT>" , False ) ) <EOL> elif not isinstance ( self . _read_delimiter , ( Struct , _NetStruct ) ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> else : <EOL> self . write ( self . _read_delimiter . pack ( * data ) , <EOL> kwargs . get ( "<STR_LIT>" , False ) ) <EOL> def flush ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _closed or self . _closing : <EOL> raise RuntimeError ( "<STR_LIT>" % self ) <EOL> if not self . connected : <EOL> raise RuntimeError ( "<STR_LIT>" % self ) <EOL> if not self . _send_buffer : <EOL> return <EOL> self . _stop_waiting_for_write_event ( ) <EOL> self . _process_send_buffer ( ) <EOL> def on_ssl_handshake ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def on_ssl_handshake_error ( self , exception ) : <EOL> """<STR_LIT>""" <EOL> log . exception ( exception ) <EOL> self . close ( flush = False ) <EOL> def on_ssl_error ( self , exception ) : <EOL> """<STR_LIT>""" <EOL> log . exception ( exception ) <EOL> self . close ( flush = False ) <EOL> def _do_connect ( self , address , family , error = None ) : <EOL> """<STR_LIT>""" <EOL> if not address : <EOL> self . connecting = False <EOL> e = StreamConnectError ( * error ) <EOL> self . _safely_call ( self . on_connect_error , e ) <EOL> return <EOL> if self . _socket : <EOL> if self . _socket . family != family : <EOL> self . engine . remove_channel ( self ) <EOL> self . _socket_close ( ) <EOL> self . _closed = False <EOL> sock = socket . socket ( family , socket . SOCK_STREAM ) <EOL> self . _socket_set ( sock ) <EOL> self . engine . add_channel ( self ) <EOL> try : <EOL> connected = self . _socket_connect ( address ) <EOL> except socket . error as err : <EOL> self . close ( flush = False ) <EOL> e = StreamConnectError ( err . errno , err . strerror ) <EOL> self . _safely_call ( self . on_connect_error , e ) <EOL> return <EOL> if connected : <EOL> self . _handle_connect_event ( ) <EOL> def _handle_read_event ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . ssl_enabled and not self . _ssl_handshake_done : <EOL> self . _ssl_do_handshake ( ) <EOL> return <EOL> while True : <EOL> try : <EOL> data = self . _socket_recv ( ) <EOL> except socket . error as err : <EOL> self . _safely_call ( self . on_read_error , err ) <EOL> return <EOL> if not data : <EOL> break <EOL> else : <EOL> self . _recv_buffer += data <EOL> if len ( self . _recv_buffer ) > self . _recv_buffer_size_limit : <EOL> self . _process_recv_buffer ( ) <EOL> if len ( self . _recv_buffer ) > self . _recv_buffer_size_limit : <EOL> e = StreamBufferOverflow ( "<STR_LIT>" % self ) <EOL> self . _safely_call ( self . on_overflow_error , e ) <EOL> return <EOL> self . _process_recv_buffer ( ) <EOL> if data is None : <EOL> self . close ( flush = False ) <EOL> def _handle_write_event ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . ssl_enabled and not self . _ssl_handshake_done : <EOL> self . _ssl_do_handshake ( ) <EOL> return <EOL> if not self . connected : <EOL> self . _handle_connect_event ( ) <EOL> if not self . _send_buffer : <EOL> return <EOL> self . _process_send_buffer ( ) <EOL> def _handle_error_event ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . connecting : <EOL> self . _handle_connect_event ( ) <EOL> else : <EOL> _Channel . _handle_error_event ( self ) <EOL> def _handle_connect_event ( self ) : <EOL> """<STR_LIT>""" <EOL> self . connecting = False <EOL> err , errstr = self . _get_socket_error ( ) <EOL> if err == <NUM_LIT:0> : <EOL> self . connected = True <EOL> if self . _ssl_enabling : <EOL> self . _ssl_call_on_connect = True <EOL> self . _process_send_buffer ( ) <EOL> else : <EOL> self . _safely_call ( self . on_connect ) <EOL> else : <EOL> e = StreamConnectError ( err , errstr ) <EOL> self . _safely_call ( self . on_connect_error , e ) <EOL> def _process_recv_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> while self . _recv_buffer : <EOL> delimiter = self . read_delimiter <EOL> if delimiter is None : <EOL> data = self . _recv_buffer <EOL> self . _recv_buffer = "<STR_LIT>" <EOL> self . _safely_call ( self . on_read , data ) <EOL> elif isinstance ( delimiter , ( int , long ) ) : <EOL> if len ( self . _recv_buffer ) < delimiter : <EOL> break <EOL> data = self . _recv_buffer [ : delimiter ] <EOL> self . _recv_buffer = self . _recv_buffer [ delimiter : ] <EOL> self . _safely_call ( self . on_read , data ) <EOL> elif isinstance ( delimiter , basestring ) : <EOL> mark = self . _recv_buffer . find ( delimiter ) <EOL> if mark == - <NUM_LIT:1> : <EOL> break <EOL> data = self . _recv_buffer [ : mark ] <EOL> self . _recv_buffer = self . _recv_buffer [ mark + len ( delimiter ) : ] <EOL> self . _safely_call ( self . on_read , data ) <EOL> elif isinstance ( delimiter , Struct ) : <EOL> if len ( self . _recv_buffer ) < delimiter . size : <EOL> break <EOL> data = self . _recv_buffer [ : delimiter . size ] <EOL> self . _recv_buffer = self . _recv_buffer [ delimiter . size : ] <EOL> try : <EOL> data = delimiter . unpack ( data ) <EOL> except struct . error : <EOL> log . exception ( "<STR_LIT>" % self ) <EOL> self . close ( ) <EOL> break <EOL> self . _safely_call ( self . on_read , * data ) <EOL> elif isinstance ( delimiter , _NetStruct ) : <EOL> if not self . _netstruct_iter : <EOL> self . _netstruct_iter = delimiter . iter_unpack ( ) <EOL> self . _netstruct_needed = next ( self . _netstruct_iter ) <EOL> if len ( self . _recv_buffer ) < self . _netstruct_needed : <EOL> break <EOL> data = self . _netstruct_iter . send ( <EOL> self . _recv_buffer [ : self . _netstruct_needed ] ) <EOL> self . _recv_buffer = self . _recv_buffer [ self . _netstruct_needed : ] <EOL> if isinstance ( data , ( int , long ) ) : <EOL> self . _netstruct_needed = data <EOL> continue <EOL> self . _netstruct_needed = None <EOL> self . _netstruct_iter = None <EOL> self . _safely_call ( self . on_read , * data ) <EOL> elif isinstance ( delimiter , RegexType ) : <EOL> if self . regex_search : <EOL> match = delimiter . search ( self . _recv_buffer ) <EOL> if not match : <EOL> break <EOL> data = self . _recv_buffer [ : match . start ( ) ] <EOL> self . _recv_buffer = self . _recv_buffer [ match . end ( ) : ] <EOL> else : <EOL> data = delimiter . match ( self . _recv_buffer ) <EOL> if not data : <EOL> break <EOL> self . _recv_buffer = self . _recv_buffer [ data . end ( ) : ] <EOL> self . _safely_call ( self . on_read , data ) <EOL> else : <EOL> err = InvalidReadDelimiterError ( "<STR_LIT>" % self ) <EOL> self . _safely_call ( self . on_error , err ) <EOL> break <EOL> if self . _closed or not self . connected : <EOL> break <EOL> def _process_send_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> while self . _send_buffer : <EOL> data_type , data = self . _send_buffer . pop ( <NUM_LIT:0> ) <EOL> if data_type == Stream . SEND_STRING : <EOL> bytes_sent = self . _process_send_string ( data ) <EOL> elif data_type == Stream . SEND_FILE : <EOL> bytes_sent = self . _process_send_file ( * data ) <EOL> elif data_type == Stream . SEND_SSL_HANDSHAKE : <EOL> bytes_sent = self . _process_send_ssl_handshake ( data ) <EOL> if bytes_sent == <NUM_LIT:0> : <EOL> break <EOL> if not self . _closed and not self . _send_buffer : <EOL> self . _safely_call ( self . on_write ) <EOL> if self . _closing : <EOL> self . close ( flush = False ) <EOL> def _process_send_string ( self , data ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> bytes_sent = self . _socket_send ( data ) <EOL> except socket . error as err : <EOL> self . _safely_call ( self . on_write_error , err ) <EOL> return <NUM_LIT:0> <EOL> if len ( data ) > bytes_sent : <EOL> self . _send_buffer . insert ( <NUM_LIT:0> , ( Stream . SEND_STRING , data [ bytes_sent : ] ) ) <EOL> return bytes_sent <EOL> def _process_send_file ( self , sfile , offset , nbytes ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> bytes_sent = self . _socket_sendfile ( sfile , offset , nbytes ) <EOL> except socket . error as err : <EOL> self . _safely_call ( self . on_write_error , err ) <EOL> return <NUM_LIT:0> <EOL> offset += bytes_sent <EOL> if nbytes > <NUM_LIT:0> : <EOL> if nbytes - bytes_sent > <NUM_LIT:0> : <EOL> nbytes -= bytes_sent <EOL> else : <EOL> return bytes_sent <EOL> if os . fstat ( sfile . fileno ( ) ) . st_size - offset <= <NUM_LIT:0> : <EOL> return bytes_sent <EOL> self . _send_buffer . insert ( <NUM_LIT:0> , ( Stream . SEND_FILE , ( sfile , offset , nbytes ) ) ) <EOL> return bytes_sent <EOL> def _process_send_ssl_handshake ( self , ssl_options ) : <EOL> """<STR_LIT>""" <EOL> self . _ssl_enabling = False <EOL> if not self . _ssl_socket_wrapped : <EOL> try : <EOL> self . _socket = ssl . wrap_socket ( self . _socket , ** ssl_options ) <EOL> except ssl . SSLError as err : <EOL> self . _ssl_enabling = True <EOL> self . _safely_call ( self . on_ssl_error , err ) <EOL> return <NUM_LIT:0> <EOL> else : <EOL> self . _ssl_socket_wrapped = True <EOL> self . ssl_enabled = True <EOL> try : <EOL> bytes_sent = self . _ssl_do_handshake ( ) <EOL> except Exception as err : <EOL> self . _safely_call ( self . on_ssl_handshake_error , err ) <EOL> return <NUM_LIT:0> <EOL> return bytes_sent <EOL> def _socket_recv ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return _Channel . _socket_recv ( self ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_READ : <EOL> return '<STR_LIT>' <EOL> else : <EOL> raise <EOL> def _socket_send ( self , data ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> bytes_sent = _Channel . _socket_send ( self , data ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_WRITE : <EOL> self . _start_waiting_for_write_event ( ) <EOL> return <NUM_LIT:0> <EOL> else : <EOL> raise <EOL> if self . ssl_enabled and bytes_sent == <NUM_LIT:0> : <EOL> self . _start_waiting_for_write_event ( ) <EOL> return bytes_sent <EOL> def _socket_sendfile ( self , sfile , offset , nbytes ) : <EOL> """<STR_LIT>""" <EOL> return _Channel . _socket_sendfile ( self , sfile , offset , nbytes , self . ssl_enabled ) <EOL> def _ssl_do_handshake ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _socket . do_handshake ( ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_READ : <EOL> return <NUM_LIT:0> <EOL> elif err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_WRITE : <EOL> self . _start_waiting_for_write_event ( ) <EOL> return <NUM_LIT:0> <EOL> elif err . args [ <NUM_LIT:0> ] in ( ssl . SSL_ERROR_EOF , ssl . SSL_ERROR_ZERO_RETURN ) : <EOL> self . close ( flush = False ) <EOL> return <NUM_LIT:0> <EOL> elif err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_SSL : <EOL> self . _safely_call ( self . on_ssl_handshake_error , err ) <EOL> return <NUM_LIT:0> <EOL> else : <EOL> raise <EOL> except socket . error as err : <EOL> if err . args [ <NUM_LIT:0> ] in ( errno . ECONNRESET , errno . EPIPE ) : <EOL> self . close ( flush = False ) <EOL> return <NUM_LIT:0> <EOL> else : <EOL> raise <EOL> else : <EOL> self . _ssl_handshake_done = True <EOL> self . _safely_call ( self . on_ssl_handshake ) <EOL> if self . _ssl_call_on_connect : <EOL> self . _safely_call ( self . on_connect ) <EOL> return None <EOL> class StreamBufferOverflow ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class StreamConnectError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class InvalidReadDelimiterError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass </s>
