<s> import sys <EOL> import socket <EOL> import select <EOL> import collections <EOL> import json <EOL> import errno <EOL> import os . path <EOL> try : <EOL> import ssl <EOL> assert ssl <EOL> except ImportError : <EOL> ssl = False <EOL> try : <EOL> from ... import editor <EOL> from . . import api , cert , msg , shared as G , utils <EOL> from . . exc_fmt import str_e <EOL> from . import base , proxy <EOL> assert cert and G and msg and proxy and utils <EOL> except ( ImportError , ValueError ) : <EOL> from floo import editor <EOL> from floo . common import api , cert , msg , shared as G , utils <EOL> from floo . common . exc_fmt import str_e <EOL> import base <EOL> import proxy <EOL> try : <EOL> connect_errno = ( errno . WSAEWOULDBLOCK , errno . WSAEALREADY , errno . WSAEINVAL ) <EOL> iscon_errno = errno . WSAEISCONN <EOL> write_again_errno = ( errno . EWOULDBLOCK , errno . EAGAIN ) + connect_errno <EOL> except Exception : <EOL> connect_errno = ( errno . EINPROGRESS , errno . EALREADY ) <EOL> iscon_errno = errno . EISCONN <EOL> write_again_errno = ( errno . EWOULDBLOCK , errno . EAGAIN ) + connect_errno <EOL> PY2 = sys . version_info < ( <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> def sock_debug ( * args , ** kwargs ) : <EOL> if G . SOCK_DEBUG : <EOL> msg . log ( * args , ** kwargs ) <EOL> class FlooProtocol ( base . BaseProtocol ) : <EOL> '''<STR_LIT>''' <EOL> MAX_RETRIES = <NUM_LIT:12> <EOL> INITIAL_RECONNECT_DELAY = <NUM_LIT> <EOL> def __init__ ( self , host , port , secure = True ) : <EOL> super ( FlooProtocol , self ) . __init__ ( host , port , secure ) <EOL> self . _handling = False <EOL> self . connected = False <EOL> self . _needs_handshake = bool ( secure ) <EOL> self . _sock = None <EOL> self . _q = collections . deque ( ) <EOL> self . _slice = bytes ( ) <EOL> self . _buf_in = bytes ( ) <EOL> self . _buf_out = bytes ( ) <EOL> self . _reconnect_delay = self . INITIAL_RECONNECT_DELAY <EOL> self . _retries = self . MAX_RETRIES <EOL> self . _empty_reads = <NUM_LIT:0> <EOL> self . _reconnect_timeout = None <EOL> self . _cert_path = os . path . join ( G . BASE_DIR , '<STR_LIT>' ) <EOL> self . req_id = <NUM_LIT:0> <EOL> self . _host = host <EOL> self . _port = port <EOL> self . _secure = secure <EOL> self . _proc = None <EOL> self . proxy = False <EOL> if secure and ssl is False : <EOL> self . proxy = True <EOL> self . _host = '<STR_LIT:127.0.0.1>' <EOL> self . _port = None <EOL> self . _secure = False <EOL> def start_proxy ( self , host , port ) : <EOL> if G . PROXY_PORT : <EOL> self . _port = int ( G . PROXY_PORT ) <EOL> msg . log ( '<STR_LIT>' % self . _port ) <EOL> return <EOL> args = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' % host , '<STR_LIT>' % str ( port ) , '<STR_LIT>' % str ( bool ( self . secure ) ) ) <EOL> self . _proc = proxy . ProxyProtocol ( ) <EOL> self . _proc . once ( '<STR_LIT>' , self . reconnect ) <EOL> self . _port = self . _proc . connect ( args ) <EOL> return self . _port <EOL> def _handle ( self , data ) : <EOL> self . _buf_in += data <EOL> if self . _handling : <EOL> return <EOL> self . _handling = True <EOL> while True : <EOL> before , sep , after = self . _buf_in . partition ( b'<STR_LIT:\n>' ) <EOL> if not sep : <EOL> break <EOL> try : <EOL> before = before . decode ( '<STR_LIT:utf-8>' , '<STR_LIT:ignore>' ) <EOL> data = json . loads ( before ) <EOL> except Exception as e : <EOL> msg . error ( '<STR_LIT>' , str_e ( e ) ) <EOL> msg . error ( '<STR_LIT>' , before ) <EOL> self . _buf_in = after <EOL> continue <EOL> name = data . get ( '<STR_LIT:name>' ) <EOL> self . _buf_in = after <EOL> try : <EOL> msg . debug ( '<STR_LIT>' + ( name or '<STR_LIT>' ) ) <EOL> self . emit ( '<STR_LIT:data>' , name , data ) <EOL> except Exception as e : <EOL> api . send_error ( '<STR_LIT>' % name , str_e ( e ) ) <EOL> if name == '<STR_LIT>' : <EOL> editor . error_message ( '<STR_LIT>' % str_e ( e ) ) <EOL> self . stop ( ) <EOL> self . _handling = False <EOL> def _connect ( self , host , port , attempts = <NUM_LIT:0> ) : <EOL> if attempts > ( self . proxy and <NUM_LIT> or <NUM_LIT> ) : <EOL> msg . error ( '<STR_LIT>' ) <EOL> return self . reconnect ( ) <EOL> if not self . _sock : <EOL> msg . debug ( '<STR_LIT>' ) <EOL> return <EOL> try : <EOL> self . _sock . connect ( ( host , port ) ) <EOL> select . select ( [ self . _sock ] , [ self . _sock ] , [ ] , <NUM_LIT:0> ) <EOL> except socket . error as e : <EOL> if e . errno == iscon_errno : <EOL> pass <EOL> elif e . errno in connect_errno : <EOL> msg . debug ( '<STR_LIT>' , str_e ( e ) ) <EOL> return utils . set_timeout ( self . _connect , <NUM_LIT:20> , host , port , attempts + <NUM_LIT:1> ) <EOL> else : <EOL> msg . error ( '<STR_LIT>' , str_e ( e ) ) <EOL> return self . reconnect ( ) <EOL> if self . _secure : <EOL> sock_debug ( '<STR_LIT>' ) <EOL> self . _sock = ssl . wrap_socket ( self . _sock , ca_certs = self . _cert_path , cert_reqs = ssl . CERT_REQUIRED , do_handshake_on_connect = False ) <EOL> self . _q . clear ( ) <EOL> self . _buf_out = bytes ( ) <EOL> self . emit ( '<STR_LIT>' ) <EOL> self . connected = True <EOL> def __len__ ( self ) : <EOL> return len ( self . _q ) <EOL> def fileno ( self ) : <EOL> return self . _sock and self . _sock . fileno ( ) <EOL> def fd_set ( self , readable , writeable , errorable ) : <EOL> if not self . connected : <EOL> return <EOL> fileno = self . fileno ( ) <EOL> errorable . append ( fileno ) <EOL> if self . _needs_handshake : <EOL> return writeable . append ( fileno ) <EOL> elif len ( self ) > <NUM_LIT:0> or self . _buf_out : <EOL> writeable . append ( fileno ) <EOL> readable . append ( fileno ) <EOL> def connect ( self , conn = None ) : <EOL> utils . cancel_timeout ( self . _reconnect_timeout ) <EOL> self . _reconnect_timeout = None <EOL> self . cleanup ( ) <EOL> host = self . _host <EOL> port = self . _port <EOL> self . _empty_selects = <NUM_LIT:0> <EOL> G . OUTBOUND_FILTERING = G . OUTBOUND_FILTERING and self . host == '<STR_LIT>' <EOL> if self . proxy : <EOL> if G . OUTBOUND_FILTERING : <EOL> port = self . start_proxy ( G . OUTBOUND_FILTER_PROXY_HOST , G . OUTBOUND_FILTER_PROXY_PORT ) <EOL> else : <EOL> port = self . start_proxy ( self . host , self . port ) <EOL> elif G . OUTBOUND_FILTERING : <EOL> host = G . OUTBOUND_FILTER_PROXY_HOST <EOL> port = G . OUTBOUND_FILTER_PROXY_PORT <EOL> self . _sock = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> self . _sock . setblocking ( False ) <EOL> if self . _secure : <EOL> with open ( self . _cert_path , '<STR_LIT:wb>' ) as cert_fd : <EOL> cert_fd . write ( cert . CA_CERT . encode ( '<STR_LIT:utf-8>' ) ) <EOL> conn_msg = '<STR_LIT>' % ( self . host , self . port ) <EOL> if self . port != self . _port or self . host != self . _host : <EOL> conn_msg += '<STR_LIT>' % ( self . _host , self . _port ) <EOL> if host != self . _host : <EOL> conn_msg += '<STR_LIT>' % ( host , port ) <EOL> msg . log ( conn_msg ) <EOL> editor . status_message ( conn_msg ) <EOL> self . _connect ( host , port ) <EOL> def cleanup ( self , * args , ** kwargs ) : <EOL> try : <EOL> self . _sock . shutdown ( <NUM_LIT:2> ) <EOL> except Exception : <EOL> pass <EOL> try : <EOL> self . _sock . close ( ) <EOL> except Exception : <EOL> pass <EOL> try : <EOL> self . _proc . cleanup ( ) <EOL> except Exception : <EOL> pass <EOL> self . _slice = bytes ( ) <EOL> self . _buf_in = bytes ( ) <EOL> self . _buf_out = bytes ( ) <EOL> self . _sock = None <EOL> self . _needs_handshake = self . _secure <EOL> self . connected = False <EOL> self . _proc = None <EOL> self . emit ( '<STR_LIT>' ) <EOL> def _do_ssl_handshake ( self ) : <EOL> try : <EOL> sock_debug ( '<STR_LIT>' ) <EOL> self . _sock . do_handshake ( ) <EOL> except ssl . SSLError as e : <EOL> sock_debug ( '<STR_LIT>' ) <EOL> if e . args [ <NUM_LIT:0> ] in [ ssl . SSL_ERROR_WANT_READ , ssl . SSL_ERROR_WANT_WRITE ] : <EOL> return False <EOL> self . stop ( ) <EOL> editor . error_message ( '<STR_LIT>' % str ( e ) ) <EOL> sock_debug ( '<STR_LIT>' % '<STR_LIT>' . join ( [ str ( a ) for a in e . args ] ) ) <EOL> except Exception as e : <EOL> msg . error ( '<STR_LIT>' , str_e ( e ) ) <EOL> else : <EOL> sock_debug ( '<STR_LIT>' ) <EOL> self . _needs_handshake = False <EOL> editor . status_message ( '<STR_LIT>' % ( self . host , self . port ) ) <EOL> return True <EOL> self . reconnect ( ) <EOL> return False <EOL> def write ( self ) : <EOL> sock_debug ( '<STR_LIT>' ) <EOL> if self . _needs_handshake and not self . _do_ssl_handshake ( ) : <EOL> return <EOL> total = <NUM_LIT:0> <EOL> if not self . _slice : <EOL> self . _slice = self . _buf_out [ total : total + <NUM_LIT> ] <EOL> try : <EOL> while True : <EOL> if total < len ( self . _buf_out ) or self . _slice : <EOL> sent = self . _sock . send ( self . _slice ) <EOL> sock_debug ( '<STR_LIT>' % ( sent , self . _slice [ - <NUM_LIT:10> : ] ) ) <EOL> if not sent : <EOL> raise IndexError ( '<STR_LIT>' ) <EOL> total += sent <EOL> self . _slice = self . _buf_out [ total : total + <NUM_LIT> ] <EOL> else : <EOL> self . _buf_out = self . _q . popleft ( ) . encode ( '<STR_LIT:utf-8>' ) <EOL> total = <NUM_LIT:0> <EOL> self . _slice = self . _buf_out [ total : total + <NUM_LIT> ] <EOL> except IndexError : <EOL> pass <EOL> except socket . error as e : <EOL> if e . errno not in write_again_errno : <EOL> raise <EOL> self . _buf_out = self . _buf_out [ total : ] <EOL> sock_debug ( '<STR_LIT>' ) <EOL> def read ( self ) : <EOL> sock_debug ( '<STR_LIT>' ) <EOL> if self . _needs_handshake and not self . _do_ssl_handshake ( ) : <EOL> return <EOL> buf = '<STR_LIT>' . encode ( '<STR_LIT:utf-8>' ) <EOL> while True : <EOL> try : <EOL> d = self . _sock . recv ( <NUM_LIT> ) <EOL> if not d : <EOL> break <EOL> buf += d <EOL> if G . SOCK_SINGLE_READ : <EOL> break <EOL> except AttributeError : <EOL> sock_debug ( '<STR_LIT>' ) <EOL> return self . reconnect ( ) <EOL> except ( socket . error , TypeError ) as e : <EOL> sock_debug ( '<STR_LIT>' , e ) <EOL> break <EOL> if buf : <EOL> self . _empty_reads = <NUM_LIT:0> <EOL> return self . _handle ( buf ) <EOL> self . _empty_reads += <NUM_LIT:1> <EOL> if self . _empty_reads > ( <NUM_LIT> / G . TICK_TIME ) : <EOL> msg . error ( '<STR_LIT>' . format ( self . _empty_reads ) ) <EOL> return self . reconnect ( ) <EOL> def error ( self ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def stop ( self ) : <EOL> self . _retries = - <NUM_LIT:1> <EOL> utils . cancel_timeout ( self . _reconnect_timeout ) <EOL> self . _reconnect_timeout = None <EOL> self . cleanup ( ) <EOL> self . emit ( '<STR_LIT>' ) <EOL> msg . log ( '<STR_LIT>' ) <EOL> def reconnect ( self ) : <EOL> if self . _reconnect_timeout : <EOL> return <EOL> self . cleanup ( ) <EOL> self . _reconnect_delay = min ( <NUM_LIT> , int ( <NUM_LIT> * self . _reconnect_delay ) ) <EOL> if self . _retries > <NUM_LIT:0> : <EOL> msg . log ( '<STR_LIT>' % self . _reconnect_delay ) <EOL> self . _reconnect_timeout = utils . set_timeout ( self . connect , self . _reconnect_delay ) <EOL> elif self . _retries == <NUM_LIT:0> : <EOL> editor . error_message ( '<STR_LIT>' ) <EOL> G . OUTBOUND_FILTERING = self . host == '<STR_LIT>' and self . _retries % <NUM_LIT:4> == <NUM_LIT:0> <EOL> self . _retries -= <NUM_LIT:1> <EOL> def reset_retries ( self ) : <EOL> self . _reconnect_delay = self . INITIAL_RECONNECT_DELAY <EOL> self . _retries = self . MAX_RETRIES <EOL> def put ( self , item ) : <EOL> if not item : <EOL> return <EOL> self . req_id += <NUM_LIT:1> <EOL> item [ '<STR_LIT>' ] = self . req_id <EOL> msg . debug ( '<STR_LIT>' , item . get ( '<STR_LIT:name>' , '<STR_LIT>' ) , <EOL> '<STR_LIT>' , self . req_id , <EOL> '<STR_LIT>' , len ( self ) ) <EOL> self . _q . append ( json . dumps ( item ) + '<STR_LIT:\n>' ) <EOL> return self . req_id </s>
