<s> """<STR_LIT>""" <EOL> import errno <EOL> from io import BytesIO <EOL> import logging <EOL> import math <EOL> import random <EOL> import re <EOL> import socket <EOL> import sys <EOL> import threading <EOL> import time <EOL> import warnings <EOL> try : <EOL> import ssl <EOL> from ssl import SSLError <EOL> DEFAULT_SSL_VERSION = ssl . PROTOCOL_TLSv1 <EOL> except ( ImportError , AttributeError ) : <EOL> ssl = None <EOL> class SSLError ( object ) : <EOL> pass <EOL> DEFAULT_SSL_VERSION = None <EOL> try : <EOL> from socket import SOL_SOCKET , SO_KEEPALIVE <EOL> from socket import SOL_TCP , TCP_KEEPIDLE , TCP_KEEPINTVL , TCP_KEEPCNT <EOL> LINUX_KEEPALIVE_AVAIL = True <EOL> except ImportError : <EOL> LINUX_KEEPALIVE_AVAIL = False <EOL> from stomp . backward import decode , encode , get_errno , monotonic , pack <EOL> from stomp . backwardsock import get_socket <EOL> import stomp . exception as exception <EOL> import stomp . listener <EOL> import stomp . utils as utils <EOL> log = logging . getLogger ( '<STR_LIT>' ) <EOL> class BaseTransport ( stomp . listener . Publisher ) : <EOL> """<STR_LIT>""" <EOL> __content_length_re = re . compile ( b'<STR_LIT>' , re . MULTILINE ) <EOL> def __init__ ( self , wait_on_receipt , auto_decode = True ) : <EOL> self . __recvbuf = b'<STR_LIT>' <EOL> self . listeners = { } <EOL> self . running = False <EOL> self . blocking = None <EOL> self . connected = False <EOL> self . connection_error = False <EOL> self . __receipts = { } <EOL> self . __wait_on_receipt = wait_on_receipt <EOL> self . current_host_and_port = None <EOL> self . __disconnect_receipt = None <EOL> self . create_thread_fc = utils . default_create_thread <EOL> self . __receiver_thread_exit_condition = threading . Condition ( ) <EOL> self . __receiver_thread_exited = False <EOL> self . __send_wait_condition = threading . Condition ( ) <EOL> self . __connect_wait_condition = threading . Condition ( ) <EOL> self . __auto_decode = auto_decode <EOL> def override_threading ( self , create_thread_fc ) : <EOL> """<STR_LIT>""" <EOL> self . create_thread_fc = create_thread_fc <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> self . running = True <EOL> self . attempt_connection ( ) <EOL> receiver_thread = self . create_thread_fc ( self . __receiver_loop ) <EOL> receiver_thread . name = "<STR_LIT>" % getattr ( receiver_thread , "<STR_LIT:name>" , "<STR_LIT>" ) <EOL> self . notify ( '<STR_LIT>' ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . __receiver_thread_exit_condition : <EOL> while not self . __receiver_thread_exited : <EOL> self . __receiver_thread_exit_condition . wait ( ) <EOL> def is_connected ( self ) : <EOL> return self . connected <EOL> def set_connected ( self , connected ) : <EOL> with self . __connect_wait_condition : <EOL> self . connected = connected <EOL> if connected : <EOL> self . __connect_wait_condition . notify ( ) <EOL> def set_listener ( self , name , listener ) : <EOL> """<STR_LIT>""" <EOL> self . listeners [ name ] = listener <EOL> def remove_listener ( self , name ) : <EOL> """<STR_LIT>""" <EOL> del self . listeners [ name ] <EOL> def get_listener ( self , name ) : <EOL> """<STR_LIT>""" <EOL> return self . listeners . get ( name ) <EOL> def process_frame ( self , f , frame_str ) : <EOL> frame_type = f . cmd . lower ( ) <EOL> if frame_type in [ '<STR_LIT>' , '<STR_LIT:message>' , '<STR_LIT>' , '<STR_LIT:error>' , '<STR_LIT>' ] : <EOL> if frame_type == '<STR_LIT:message>' : <EOL> ( f . headers , f . body ) = self . notify ( '<STR_LIT>' , f . headers , f . body ) <EOL> if log . isEnabledFor ( logging . DEBUG ) : <EOL> log . debug ( "<STR_LIT>" , f . cmd , f . headers , f . body ) <EOL> else : <EOL> log . info ( "<STR_LIT>" , f . cmd , f . headers , utils . length ( f . body ) ) <EOL> self . notify ( frame_type , f . headers , f . body ) <EOL> else : <EOL> log . warning ( "<STR_LIT>" , frame_type , utils . length ( frame_str ) ) <EOL> def notify ( self , frame_type , headers = None , body = None ) : <EOL> """<STR_LIT>""" <EOL> if frame_type == '<STR_LIT>' : <EOL> receipt = headers [ '<STR_LIT>' ] <EOL> with self . __send_wait_condition : <EOL> self . __receipts [ receipt ] = None <EOL> self . __send_wait_condition . notify ( ) <EOL> if receipt == self . __disconnect_receipt : <EOL> self . disconnect_socket ( ) <EOL> elif frame_type == '<STR_LIT>' : <EOL> self . set_connected ( True ) <EOL> elif frame_type == '<STR_LIT>' : <EOL> self . set_connected ( False ) <EOL> for listener in self . listeners . values ( ) : <EOL> if not listener : <EOL> continue <EOL> notify_func = getattr ( listener , '<STR_LIT>' % frame_type , None ) <EOL> if not notify_func : <EOL> log . debug ( "<STR_LIT>" , listener , frame_type ) <EOL> continue <EOL> if frame_type in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> notify_func ( ) <EOL> continue <EOL> if frame_type == '<STR_LIT>' : <EOL> notify_func ( self . current_host_and_port ) <EOL> continue <EOL> if frame_type == '<STR_LIT:error>' and not self . connected : <EOL> with self . __connect_wait_condition : <EOL> self . connection_error = True <EOL> self . __connect_wait_condition . notify ( ) <EOL> rtn = notify_func ( headers , body ) <EOL> if rtn : <EOL> ( headers , body ) = rtn <EOL> return ( headers , body ) <EOL> def transmit ( self , frame ) : <EOL> """<STR_LIT>""" <EOL> for listener in self . listeners . values ( ) : <EOL> if not listener : <EOL> continue <EOL> try : <EOL> listener . on_send ( frame ) <EOL> except AttributeError : <EOL> continue <EOL> lines = utils . convert_frame_to_lines ( frame ) <EOL> packed_frame = pack ( lines ) <EOL> if log . isEnabledFor ( logging . DEBUG ) : <EOL> log . debug ( "<STR_LIT>" , lines ) <EOL> else : <EOL> log . info ( "<STR_LIT>" , frame . cmd , frame . headers ) <EOL> self . send ( encode ( packed_frame ) ) <EOL> def send ( self , encoded_frame ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def receive ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def cleanup ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def attempt_connection ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def disconnect_socket ( self ) : <EOL> """<STR_LIT>""" <EOL> def wait_for_connection ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if timeout is not None : <EOL> wait_time = timeout / <NUM_LIT> <EOL> else : <EOL> wait_time = None <EOL> with self . __connect_wait_condition : <EOL> while not self . is_connected ( ) and not self . connection_error : <EOL> self . __connect_wait_condition . wait ( wait_time ) <EOL> def __receiver_loop ( self ) : <EOL> """<STR_LIT>""" <EOL> log . info ( "<STR_LIT>" ) <EOL> try : <EOL> while self . running : <EOL> try : <EOL> while self . running : <EOL> frames = self . __read ( ) <EOL> for frame in frames : <EOL> f = utils . parse_frame ( frame ) <EOL> if self . __auto_decode : <EOL> f . body = decode ( f . body ) <EOL> self . process_frame ( f , frame ) <EOL> except exception . ConnectionClosedException : <EOL> if self . running : <EOL> self . notify ( '<STR_LIT>' ) <EOL> self . __recvbuf = b'<STR_LIT>' <EOL> self . running = False <EOL> break <EOL> finally : <EOL> self . cleanup ( ) <EOL> finally : <EOL> with self . __receiver_thread_exit_condition : <EOL> self . __receiver_thread_exited = True <EOL> self . __receiver_thread_exit_condition . notifyAll ( ) <EOL> log . info ( "<STR_LIT>" ) <EOL> def __read ( self ) : <EOL> """<STR_LIT>""" <EOL> fastbuf = BytesIO ( ) <EOL> while self . running : <EOL> try : <EOL> try : <EOL> c = self . receive ( ) <EOL> except exception . InterruptedException : <EOL> log . debug ( "<STR_LIT>" ) <EOL> continue <EOL> except Exception : <EOL> c = b'<STR_LIT>' <EOL> if len ( c ) == <NUM_LIT:0> : <EOL> raise exception . ConnectionClosedException ( ) <EOL> fastbuf . write ( c ) <EOL> if b'<STR_LIT:\x00>' in c : <EOL> break <EOL> elif c == b'<STR_LIT>' : <EOL> return [ c , ] <EOL> self . __recvbuf += fastbuf . getvalue ( ) <EOL> fastbuf . close ( ) <EOL> result = [ ] <EOL> if self . __recvbuf and self . running : <EOL> while True : <EOL> pos = self . __recvbuf . find ( b'<STR_LIT:\x00>' ) <EOL> if pos >= <NUM_LIT:0> : <EOL> frame = self . __recvbuf [ <NUM_LIT:0> : pos ] <EOL> preamble_end_match = utils . PREAMBLE_END_RE . search ( frame ) <EOL> if preamble_end_match : <EOL> preamble_end = preamble_end_match . start ( ) <EOL> content_length_match = BaseTransport . __content_length_re . search ( frame [ <NUM_LIT:0> : preamble_end ] ) <EOL> if content_length_match : <EOL> content_length = int ( content_length_match . group ( '<STR_LIT:value>' ) ) <EOL> content_offset = preamble_end_match . end ( ) <EOL> frame_size = content_offset + content_length <EOL> if frame_size > len ( frame ) : <EOL> if frame_size < len ( self . __recvbuf ) : <EOL> pos = frame_size <EOL> frame = self . __recvbuf [ <NUM_LIT:0> : pos ] <EOL> else : <EOL> break <EOL> result . append ( frame ) <EOL> self . __recvbuf = self . __recvbuf [ pos + <NUM_LIT:1> : ] <EOL> else : <EOL> break <EOL> return result <EOL> class Transport ( BaseTransport ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> host_and_ports = None , <EOL> prefer_localhost = True , <EOL> try_loopback_connect = True , <EOL> reconnect_sleep_initial = <NUM_LIT:0.1> , <EOL> reconnect_sleep_increase = <NUM_LIT:0.5> , <EOL> reconnect_sleep_jitter = <NUM_LIT:0.1> , <EOL> reconnect_sleep_max = <NUM_LIT> , <EOL> reconnect_attempts_max = <NUM_LIT:3> , <EOL> use_ssl = False , <EOL> ssl_key_file = None , <EOL> ssl_cert_file = None , <EOL> ssl_ca_certs = None , <EOL> ssl_cert_validator = None , <EOL> wait_on_receipt = False , <EOL> ssl_version = None , <EOL> timeout = None , <EOL> keepalive = None , <EOL> vhost = None , <EOL> auto_decode = True <EOL> ) : <EOL> BaseTransport . __init__ ( self , wait_on_receipt , auto_decode ) <EOL> if host_and_ports is None : <EOL> host_and_ports = [ ( '<STR_LIT:localhost>' , <NUM_LIT> ) ] <EOL> sorted_host_and_ports = [ ] <EOL> sorted_host_and_ports . extend ( host_and_ports ) <EOL> if prefer_localhost : <EOL> sorted_host_and_ports . sort ( key = utils . is_localhost ) <EOL> loopback_host_and_ports = [ ] <EOL> if try_loopback_connect : <EOL> for host_and_port in sorted_host_and_ports : <EOL> if utils . is_localhost ( host_and_port ) == <NUM_LIT:1> : <EOL> port = host_and_port [ <NUM_LIT:1> ] <EOL> if not ( ( "<STR_LIT:127.0.0.1>" , port ) in sorted_host_and_ports or ( "<STR_LIT:localhost>" , port ) in sorted_host_and_ports ) : <EOL> loopback_host_and_ports . append ( ( "<STR_LIT:127.0.0.1>" , port ) ) <EOL> self . __host_and_ports = [ ] <EOL> self . __host_and_ports . extend ( loopback_host_and_ports ) <EOL> self . __host_and_ports . extend ( sorted_host_and_ports ) <EOL> self . __reconnect_sleep_initial = reconnect_sleep_initial <EOL> self . __reconnect_sleep_increase = reconnect_sleep_increase <EOL> self . __reconnect_sleep_jitter = reconnect_sleep_jitter <EOL> self . __reconnect_sleep_max = reconnect_sleep_max <EOL> self . __reconnect_attempts_max = reconnect_attempts_max <EOL> self . __timeout = timeout <EOL> self . socket = None <EOL> self . __socket_semaphore = threading . BoundedSemaphore ( <NUM_LIT:1> ) <EOL> self . current_host_and_port = None <EOL> self . __ssl_params = { } <EOL> if use_ssl : <EOL> warnings . warn ( "<STR_LIT>" , DeprecationWarning ) <EOL> self . set_ssl ( host_and_ports , <EOL> ssl_key_file , <EOL> ssl_cert_file , <EOL> ssl_ca_certs , <EOL> ssl_cert_validator , <EOL> ssl_version ) <EOL> self . __keepalive = keepalive <EOL> self . vhost = vhost <EOL> def is_connected ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . socket is not None and self . socket . getsockname ( ) [ <NUM_LIT:1> ] != <NUM_LIT:0> and BaseTransport . is_connected ( self ) <EOL> except socket . error : <EOL> return False <EOL> def disconnect_socket ( self ) : <EOL> """<STR_LIT>""" <EOL> self . running = False <EOL> if self . socket is not None : <EOL> if self . __need_ssl ( ) : <EOL> try : <EOL> self . socket = self . socket . unwrap ( ) <EOL> except Exception : <EOL> _ , e , _ = sys . exc_info ( ) <EOL> log . warning ( e ) <EOL> elif hasattr ( socket , '<STR_LIT>' ) : <EOL> try : <EOL> self . socket . shutdown ( socket . SHUT_RDWR ) <EOL> except socket . error : <EOL> _ , e , _ = sys . exc_info ( ) <EOL> if get_errno ( e ) != errno . ENOTCONN : <EOL> log . warning ( "<STR_LIT>" , e ) <EOL> if self . socket is not None : <EOL> try : <EOL> self . socket . close ( ) <EOL> except socket . error : <EOL> _ , e , _ = sys . exc_info ( ) <EOL> log . warning ( "<STR_LIT>" , e ) <EOL> self . current_host_and_port = None <EOL> def send ( self , encoded_frame ) : <EOL> if self . socket is not None : <EOL> try : <EOL> with self . __socket_semaphore : <EOL> self . socket . sendall ( encoded_frame ) <EOL> except Exception : <EOL> _ , e , _ = sys . exc_info ( ) <EOL> log . error ( "<STR_LIT>" , exc_info = <NUM_LIT:1> ) <EOL> raise e <EOL> else : <EOL> raise exception . NotConnectedException ( ) <EOL> def receive ( self ) : <EOL> try : <EOL> return self . socket . recv ( <NUM_LIT> ) <EOL> except socket . error : <EOL> _ , e , _ = sys . exc_info ( ) <EOL> if get_errno ( e ) in ( errno . EAGAIN , errno . EINTR ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> raise exception . InterruptedException ( ) <EOL> raise <EOL> def cleanup ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . socket . close ( ) <EOL> except : <EOL> pass <EOL> self . socket = None <EOL> self . current_host_and_port = None <EOL> def __enable_keepalive ( self ) : <EOL> def try_setsockopt ( sock , name , fam , opt , val ) : <EOL> if val is None : <EOL> return True <EOL> try : <EOL> sock . setsockopt ( fam , opt , val ) <EOL> log . info ( "<STR_LIT>" , name , val ) <EOL> except : <EOL> log . error ( "<STR_LIT>" , name , val ) <EOL> return False <EOL> return True <EOL> ka = self . __keepalive <EOL> if not ka : <EOL> return <EOL> if ka is True : <EOL> ka_sig = '<STR_LIT>' <EOL> ka_args = ( ) <EOL> else : <EOL> try : <EOL> ka_sig = ka [ <NUM_LIT:0> ] <EOL> ka_args = ka [ <NUM_LIT:1> : ] <EOL> except Exception : <EOL> log . error ( "<STR_LIT>" , ka ) <EOL> return <EOL> if ka_sig == '<STR_LIT>' : <EOL> if LINUX_KEEPALIVE_AVAIL : <EOL> ka_sig = '<STR_LIT>' <EOL> ka_args = None <EOL> log . info ( "<STR_LIT>" ) <EOL> else : <EOL> log . error ( "<STR_LIT>" ) <EOL> return <EOL> if ka_sig == '<STR_LIT>' : <EOL> log . info ( "<STR_LIT>" ) <EOL> if ka_args is None : <EOL> log . info ( "<STR_LIT>" ) <EOL> ka_args = ( None , None , None ) <EOL> lka_idle , lka_intvl , lka_cnt = ka_args <EOL> if try_setsockopt ( self . socket , '<STR_LIT>' , SOL_SOCKET , SO_KEEPALIVE , <NUM_LIT:1> ) : <EOL> try_setsockopt ( self . socket , '<STR_LIT>' , SOL_TCP , TCP_KEEPIDLE , lka_idle ) <EOL> try_setsockopt ( self . socket , '<STR_LIT>' , SOL_TCP , TCP_KEEPINTVL , lka_intvl ) <EOL> try_setsockopt ( self . socket , '<STR_LIT:count>' , SOL_TCP , TCP_KEEPCNT , lka_cnt ) <EOL> else : <EOL> log . error ( "<STR_LIT>" , ka_sig ) <EOL> def attempt_connection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . connection_error = False <EOL> sleep_exp = <NUM_LIT:1> <EOL> connect_count = <NUM_LIT:0> <EOL> while self . running and self . socket is None and connect_count < self . __reconnect_attempts_max : <EOL> for host_and_port in self . __host_and_ports : <EOL> try : <EOL> log . info ( "<STR_LIT>" , host_and_port [ <NUM_LIT:0> ] , host_and_port [ <NUM_LIT:1> ] ) <EOL> self . socket = get_socket ( host_and_port [ <NUM_LIT:0> ] , host_and_port [ <NUM_LIT:1> ] , self . __timeout ) <EOL> self . __enable_keepalive ( ) <EOL> need_ssl = self . __need_ssl ( host_and_port ) <EOL> if need_ssl : <EOL> ssl_params = self . get_ssl ( host_and_port ) <EOL> if ssl_params [ '<STR_LIT>' ] : <EOL> cert_validation = ssl . CERT_REQUIRED <EOL> else : <EOL> cert_validation = ssl . CERT_NONE <EOL> self . socket = ssl . wrap_socket ( <EOL> self . socket , <EOL> keyfile = ssl_params [ '<STR_LIT>' ] , <EOL> certfile = ssl_params [ '<STR_LIT>' ] , <EOL> cert_reqs = cert_validation , <EOL> ca_certs = ssl_params [ '<STR_LIT>' ] , <EOL> ssl_version = ssl_params [ '<STR_LIT>' ] ) <EOL> self . socket . settimeout ( self . __timeout ) <EOL> if self . blocking is not None : <EOL> self . socket . setblocking ( self . blocking ) <EOL> if need_ssl and ssl_params [ '<STR_LIT>' ] : <EOL> cert = self . socket . getpeercert ( ) <EOL> ( ok , errmsg ) = ssl_params [ '<STR_LIT>' ] ( cert , host_and_port [ <NUM_LIT:0> ] ) <EOL> if not ok : <EOL> raise SSLError ( "<STR_LIT>" , errmsg ) <EOL> self . current_host_and_port = host_and_port <EOL> log . info ( "<STR_LIT>" , host_and_port [ <NUM_LIT:0> ] , host_and_port [ <NUM_LIT:1> ] ) <EOL> break <EOL> except socket . error : <EOL> self . socket = None <EOL> connect_count += <NUM_LIT:1> <EOL> log . warning ( "<STR_LIT>" , host_and_port [ <NUM_LIT:0> ] , host_and_port [ <NUM_LIT:1> ] , exc_info = <NUM_LIT:1> ) <EOL> if self . socket is None : <EOL> sleep_duration = ( min ( self . __reconnect_sleep_max , <EOL> ( ( self . __reconnect_sleep_initial / ( <NUM_LIT:1.0> + self . __reconnect_sleep_increase ) ) <EOL> * math . pow ( <NUM_LIT:1.0> + self . __reconnect_sleep_increase , sleep_exp ) ) ) <EOL> * ( <NUM_LIT:1.0> + random . random ( ) * self . __reconnect_sleep_jitter ) ) <EOL> sleep_end = monotonic ( ) + sleep_duration <EOL> log . debug ( "<STR_LIT>" , sleep_duration ) <EOL> while self . running and monotonic ( ) < sleep_end : <EOL> time . sleep ( <NUM_LIT> ) <EOL> if sleep_duration < self . __reconnect_sleep_max : <EOL> sleep_exp += <NUM_LIT:1> <EOL> if not self . socket : <EOL> raise exception . ConnectFailedException ( ) <EOL> def set_ssl ( self , <EOL> for_hosts = [ ] , <EOL> key_file = None , <EOL> cert_file = None , <EOL> ca_certs = None , <EOL> cert_validator = None , <EOL> ssl_version = DEFAULT_SSL_VERSION ) : <EOL> """<STR_LIT>""" <EOL> if not ssl : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> for host_port in for_hosts : <EOL> self . __ssl_params [ host_port ] = dict ( key_file = key_file , <EOL> cert_file = cert_file , <EOL> ca_certs = ca_certs , <EOL> cert_validator = cert_validator , <EOL> ssl_version = ssl_version ) <EOL> def __need_ssl ( self , host_and_port = None ) : <EOL> """<STR_LIT>""" <EOL> if not host_and_port : <EOL> host_and_port = self . current_host_and_port <EOL> return host_and_port in self . __ssl_params <EOL> def get_ssl ( self , host_and_port = None ) : <EOL> """<STR_LIT>""" <EOL> if not host_and_port : <EOL> host_and_port = self . current_host_and_port <EOL> return self . __ssl_params . get ( host_and_port ) </s>
