<s> import json <EOL> import Queue as queue <EOL> import socket <EOL> import select <EOL> import threading <EOL> import time <EOL> import sys <EOL> from processor import Session , Dispatcher <EOL> from utils import print_log , logger <EOL> READ_ONLY = select . POLLIN | select . POLLPRI | select . POLLHUP | select . POLLERR <EOL> READ_WRITE = READ_ONLY | select . POLLOUT <EOL> WRITE_ONLY = select . POLLOUT <EOL> TIMEOUT = <NUM_LIT:100> <EOL> import ssl <EOL> class TcpSession ( Session ) : <EOL> def __init__ ( self , dispatcher , connection , address , use_ssl , ssl_certfile , ssl_keyfile ) : <EOL> Session . __init__ ( self , dispatcher ) <EOL> self . use_ssl = use_ssl <EOL> self . raw_connection = connection <EOL> if use_ssl : <EOL> import ssl <EOL> self . _connection = ssl . wrap_socket ( <EOL> connection , <EOL> server_side = True , <EOL> certfile = ssl_certfile , <EOL> keyfile = ssl_keyfile , <EOL> ssl_version = ssl . PROTOCOL_SSLv23 , <EOL> do_handshake_on_connect = False ) <EOL> else : <EOL> self . _connection = connection <EOL> self . address = address [ <NUM_LIT:0> ] + "<STR_LIT>" % address [ <NUM_LIT:1> ] <EOL> self . name = "<STR_LIT>" if not use_ssl else "<STR_LIT>" <EOL> self . timeout = <NUM_LIT:1000> <EOL> self . dispatcher . add_session ( self ) <EOL> self . response_queue = queue . Queue ( ) <EOL> self . message = '<STR_LIT>' <EOL> self . retry_msg = '<STR_LIT>' <EOL> self . handshake = not self . use_ssl <EOL> self . mode = None <EOL> def connection ( self ) : <EOL> if self . stopped ( ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> else : <EOL> return self . _connection <EOL> def shutdown ( self ) : <EOL> try : <EOL> self . _connection . shutdown ( socket . SHUT_RDWR ) <EOL> except : <EOL> pass <EOL> self . _connection . close ( ) <EOL> def send_response ( self , response ) : <EOL> try : <EOL> msg = json . dumps ( response ) + '<STR_LIT:\n>' <EOL> except BaseException as e : <EOL> logger . error ( '<STR_LIT>' + str ( e ) ) <EOL> return <EOL> self . response_queue . put ( msg ) <EOL> def parse_message ( self ) : <EOL> message = self . message <EOL> self . time = time . time ( ) <EOL> raw_buffer = message . find ( '<STR_LIT:\n>' ) <EOL> if raw_buffer == - <NUM_LIT:1> : <EOL> return False <EOL> raw_command = message [ <NUM_LIT:0> : raw_buffer ] . strip ( ) <EOL> self . message = message [ raw_buffer + <NUM_LIT:1> : ] <EOL> return raw_command <EOL> class TcpServer ( threading . Thread ) : <EOL> def __init__ ( self , dispatcher , host , port , use_ssl , ssl_certfile , ssl_keyfile ) : <EOL> self . shared = dispatcher . shared <EOL> self . dispatcher = dispatcher . request_dispatcher <EOL> threading . Thread . __init__ ( self ) <EOL> self . daemon = True <EOL> self . host = host <EOL> self . port = port <EOL> self . lock = threading . Lock ( ) <EOL> self . use_ssl = use_ssl <EOL> self . ssl_keyfile = ssl_keyfile <EOL> self . ssl_certfile = ssl_certfile <EOL> self . fd_to_session = { } <EOL> self . buffer_size = <NUM_LIT> <EOL> def handle_command ( self , raw_command , session ) : <EOL> try : <EOL> command = json . loads ( raw_command ) <EOL> except : <EOL> session . send_response ( { "<STR_LIT:error>" : "<STR_LIT>" } ) <EOL> return True <EOL> try : <EOL> message_id = command [ '<STR_LIT:id>' ] <EOL> method = command [ '<STR_LIT>' ] <EOL> except : <EOL> session . send_response ( { "<STR_LIT:error>" : "<STR_LIT>" , "<STR_LIT>" : raw_command } ) <EOL> else : <EOL> self . dispatcher . push_request ( session , command ) <EOL> def run ( self ) : <EOL> for res in socket . getaddrinfo ( self . host , self . port , socket . AF_UNSPEC , socket . SOCK_STREAM ) : <EOL> af , socktype , proto , cannonname , sa = res <EOL> try : <EOL> sock = socket . socket ( af , socktype , proto ) <EOL> sock . setblocking ( <NUM_LIT:0> ) <EOL> sock . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> except socket . error : <EOL> sock = None <EOL> continue <EOL> try : <EOL> sock . bind ( sa ) <EOL> sock . listen ( <NUM_LIT:5> ) <EOL> except socket . error : <EOL> sock . close ( ) <EOL> sock = None <EOL> continue <EOL> break <EOL> host = sa [ <NUM_LIT:0> ] <EOL> if af == socket . AF_INET6 : <EOL> host = "<STR_LIT>" % host <EOL> if sock is None : <EOL> print_log ( "<STR_LIT>" + ( "<STR_LIT>" if self . use_ssl else "<STR_LIT>" ) + "<STR_LIT>" % ( host , self . port ) ) <EOL> return <EOL> print_log ( ( "<STR_LIT>" if self . use_ssl else "<STR_LIT>" ) + "<STR_LIT>" % ( host , self . port ) ) <EOL> sock_fd = sock . fileno ( ) <EOL> poller = select . poll ( ) <EOL> poller . register ( sock ) <EOL> def stop_session ( fd ) : <EOL> try : <EOL> poller . unregister ( fd ) <EOL> session = self . fd_to_session . pop ( fd ) <EOL> session . stop ( ) <EOL> except BaseException as e : <EOL> logger . error ( '<STR_LIT>' + str ( e ) ) <EOL> def check_do_handshake ( session ) : <EOL> if session . handshake : <EOL> return <EOL> try : <EOL> session . _connection . do_handshake ( ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_READ : <EOL> return <EOL> elif err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_WRITE : <EOL> poller . modify ( session . raw_connection , READ_WRITE ) <EOL> return <EOL> else : <EOL> raise BaseException ( str ( err ) ) <EOL> poller . modify ( session . raw_connection , READ_ONLY ) <EOL> session . handshake = True <EOL> redo = [ ] <EOL> while not self . shared . stopped ( ) : <EOL> if self . shared . paused ( ) : <EOL> sessions = self . fd_to_session . keys ( ) <EOL> if sessions : <EOL> logger . info ( "<STR_LIT>" % len ( sessions ) ) <EOL> for fd in sessions : <EOL> stop_session ( fd ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> continue <EOL> if redo : <EOL> events = redo <EOL> redo = [ ] <EOL> else : <EOL> now = time . time ( ) <EOL> for fd , session in self . fd_to_session . items ( ) : <EOL> if now - session . time > <NUM_LIT> and session . message : <EOL> cmd = session . parse_message ( ) <EOL> if not cmd : <EOL> break <EOL> if cmd == '<STR_LIT>' : <EOL> data = False <EOL> break <EOL> session . time = now <EOL> self . handle_command ( cmd , session ) <EOL> if session . response_queue . empty ( ) : <EOL> mode = READ_ONLY <EOL> elif session . response_queue . qsize ( ) < <NUM_LIT:200> : <EOL> mode = READ_WRITE <EOL> else : <EOL> mode = WRITE_ONLY <EOL> if mode != session . mode : <EOL> poller . modify ( session . raw_connection , mode ) <EOL> session . mode = mode <EOL> if now - session . time > session . timeout : <EOL> stop_session ( fd ) <EOL> events = poller . poll ( TIMEOUT ) <EOL> for fd , flag in events : <EOL> if fd == sock_fd : <EOL> if flag & ( select . POLLIN | select . POLLPRI ) : <EOL> try : <EOL> connection , address = sock . accept ( ) <EOL> session = TcpSession ( self . dispatcher , connection , address , <EOL> use_ssl = self . use_ssl , ssl_certfile = self . ssl_certfile , ssl_keyfile = self . ssl_keyfile ) <EOL> except BaseException as e : <EOL> logger . error ( "<STR_LIT>" + str ( e ) + '<STR_LIT:U+0020>' + repr ( address ) ) <EOL> connection . close ( ) <EOL> continue <EOL> connection = session . _connection <EOL> connection . setblocking ( False ) <EOL> self . fd_to_session [ connection . fileno ( ) ] = session <EOL> poller . register ( connection , READ_ONLY ) <EOL> continue <EOL> session = self . fd_to_session [ fd ] <EOL> s = session . _connection <EOL> try : <EOL> check_do_handshake ( session ) <EOL> except BaseException as e : <EOL> stop_session ( fd ) <EOL> continue <EOL> now = time . time ( ) <EOL> if now - session . time < <NUM_LIT> : <EOL> continue <EOL> if flag & ( select . POLLIN | select . POLLPRI ) : <EOL> try : <EOL> data = s . recv ( self . buffer_size ) <EOL> except ssl . SSLError as x : <EOL> if x . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_READ : <EOL> pass <EOL> elif x . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_SSL : <EOL> pass <EOL> else : <EOL> logger . error ( '<STR_LIT>' + repr ( x ) ) <EOL> continue <EOL> except socket . error as x : <EOL> if x . args [ <NUM_LIT:0> ] != <NUM_LIT> : <EOL> logger . error ( '<STR_LIT>' + repr ( x ) + '<STR_LIT>' % fd ) <EOL> stop_session ( fd ) <EOL> continue <EOL> except ValueError as e : <EOL> logger . error ( '<STR_LIT>' + str ( e ) + '<STR_LIT>' % fd ) <EOL> stop_session ( fd ) <EOL> continue <EOL> if data : <EOL> session . message += data <EOL> if len ( data ) == self . buffer_size : <EOL> redo . append ( ( fd , flag ) ) <EOL> if not data : <EOL> stop_session ( fd ) <EOL> continue <EOL> elif flag & select . POLLHUP : <EOL> print_log ( '<STR_LIT>' , session . address ) <EOL> stop_session ( fd ) <EOL> elif flag & select . POLLOUT : <EOL> if session . retry_msg : <EOL> next_msg = session . retry_msg <EOL> else : <EOL> try : <EOL> next_msg = session . response_queue . get_nowait ( ) <EOL> except queue . Empty : <EOL> continue <EOL> try : <EOL> sent = s . send ( next_msg ) <EOL> except socket . error as x : <EOL> logger . error ( "<STR_LIT>" + str ( x ) ) <EOL> stop_session ( fd ) <EOL> continue <EOL> session . retry_msg = next_msg [ sent : ] <EOL> elif flag & select . POLLERR : <EOL> print_log ( '<STR_LIT>' , session . address ) <EOL> stop_session ( fd ) <EOL> elif flag & select . POLLNVAL : <EOL> print_log ( '<STR_LIT>' , session . address ) <EOL> stop_session ( fd ) <EOL> print_log ( '<STR_LIT>' , self . shared . stopped ( ) ) </s>
