<s> """<STR_LIT>""" <EOL> from __future__ import unicode_literals <EOL> import socket <EOL> import select <EOL> import threading <EOL> import os <EOL> import fcntl <EOL> from six import int2byte , text_type , binary_type <EOL> from codecs import getincrementaldecoder <EOL> from prompt_toolkit . enums import DEFAULT_BUFFER <EOL> from prompt_toolkit . eventloop . base import EventLoop <EOL> from prompt_toolkit . interface import CommandLineInterface , Application <EOL> from prompt_toolkit . layout . screen import Size <EOL> from prompt_toolkit . shortcuts import create_prompt_application <EOL> from prompt_toolkit . terminal . vt100_input import InputStream <EOL> from prompt_toolkit . terminal . vt100_output import Vt100_Output <EOL> from . log import logger <EOL> from . protocol import IAC , DO , LINEMODE , SB , MODE , SE , WILL , ECHO , NAWS , SUPPRESS_GO_AHEAD <EOL> from . protocol import TelnetProtocolParser <EOL> from . application import TelnetApplication <EOL> __all__ = ( <EOL> '<STR_LIT>' , <EOL> ) <EOL> def _initialize_telnet ( connection ) : <EOL> logger . info ( '<STR_LIT>' ) <EOL> connection . send ( IAC + DO + LINEMODE ) <EOL> connection . send ( IAC + WILL + SUPPRESS_GO_AHEAD ) <EOL> connection . send ( IAC + SB + LINEMODE + MODE + int2byte ( <NUM_LIT:0> ) + IAC + SE ) <EOL> connection . send ( IAC + WILL + ECHO ) <EOL> connection . send ( IAC + DO + NAWS ) <EOL> class _ConnectionStdout ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , connection , encoding ) : <EOL> self . _encoding = encoding <EOL> self . _connection = connection <EOL> self . _buffer = [ ] <EOL> def write ( self , data ) : <EOL> assert isinstance ( data , text_type ) <EOL> self . _buffer . append ( data . encode ( self . _encoding ) ) <EOL> self . flush ( ) <EOL> def flush ( self ) : <EOL> try : <EOL> self . _connection . send ( b'<STR_LIT>' . join ( self . _buffer ) ) <EOL> except socket . error as e : <EOL> logger . error ( "<STR_LIT>" % e ) <EOL> self . _buffer = [ ] <EOL> class TelnetConnection ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , conn , addr , application , server , encoding ) : <EOL> assert isinstance ( addr , tuple ) <EOL> assert isinstance ( application , TelnetApplication ) <EOL> assert isinstance ( server , TelnetServer ) <EOL> assert isinstance ( encoding , text_type ) <EOL> self . conn = conn <EOL> self . addr = addr <EOL> self . application = application <EOL> self . closed = False <EOL> self . handling_command = True <EOL> self . server = server <EOL> self . encoding = encoding <EOL> self . callback = None <EOL> self . size = Size ( rows = <NUM_LIT> , columns = <NUM_LIT> ) <EOL> _initialize_telnet ( conn ) <EOL> def get_size ( ) : <EOL> return self . size <EOL> self . stdout = _ConnectionStdout ( conn , encoding = encoding ) <EOL> self . vt100_output = Vt100_Output ( self . stdout , get_size ) <EOL> self . eventloop = _TelnetEventLoopInterface ( server ) <EOL> self . set_application ( create_prompt_application ( ) ) <EOL> application . client_connected ( self ) <EOL> self . handling_command = False <EOL> self . cli . _redraw ( ) <EOL> def set_application ( self , app , callback = None ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( app , Application ) <EOL> assert callback is None or callable ( callback ) <EOL> self . cli = CommandLineInterface ( <EOL> application = app , <EOL> eventloop = self . eventloop , <EOL> output = self . vt100_output ) <EOL> self . callback = callback <EOL> cb = self . cli . create_eventloop_callbacks ( ) <EOL> inputstream = InputStream ( cb . feed_key ) <EOL> stdin_decoder_cls = getincrementaldecoder ( self . encoding ) <EOL> stdin_decoder = [ stdin_decoder_cls ( ) ] <EOL> self . cli . _is_running = True <EOL> def data_received ( data ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( data , binary_type ) <EOL> try : <EOL> result = stdin_decoder [ <NUM_LIT:0> ] . decode ( data ) <EOL> inputstream . feed ( result ) <EOL> except UnicodeDecodeError : <EOL> stdin_decoder [ <NUM_LIT:0> ] = stdin_decoder_cls ( ) <EOL> return '<STR_LIT>' <EOL> def size_received ( rows , columns ) : <EOL> """<STR_LIT>""" <EOL> self . size = Size ( rows = rows , columns = columns ) <EOL> cb . terminal_size_changed ( ) <EOL> self . parser = TelnetProtocolParser ( data_received , size_received ) <EOL> def feed ( self , data ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( data , binary_type ) <EOL> self . parser . feed ( data ) <EOL> self . cli . _redraw ( ) <EOL> if self . cli . is_returning : <EOL> try : <EOL> return_value = self . cli . return_value ( ) <EOL> except ( EOFError , KeyboardInterrupt ) as e : <EOL> logger . info ( '<STR_LIT>' , type ( e ) . __name__ ) <EOL> self . close ( ) <EOL> return <EOL> self . _handle_command ( return_value ) <EOL> def _handle_command ( self , command ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( '<STR_LIT>' , command ) <EOL> def in_executor ( ) : <EOL> self . handling_command = True <EOL> try : <EOL> if self . callback is not None : <EOL> self . callback ( self , command ) <EOL> finally : <EOL> self . server . call_from_executor ( done ) <EOL> def done ( ) : <EOL> self . handling_command = False <EOL> if not self . closed : <EOL> self . cli . reset ( ) <EOL> self . cli . buffers [ DEFAULT_BUFFER ] . reset ( ) <EOL> self . cli . renderer . request_absolute_cursor_position ( ) <EOL> self . vt100_output . flush ( ) <EOL> self . cli . _redraw ( ) <EOL> self . server . run_in_executor ( in_executor ) <EOL> def erase_screen ( self ) : <EOL> """<STR_LIT>""" <EOL> self . vt100_output . erase_screen ( ) <EOL> self . vt100_output . cursor_goto ( <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> self . vt100_output . flush ( ) <EOL> def send ( self , data ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( data , text_type ) <EOL> self . stdout . write ( data . replace ( '<STR_LIT:\n>' , '<STR_LIT:\r\n>' ) ) <EOL> self . stdout . flush ( ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . application . client_leaving ( self ) <EOL> self . conn . close ( ) <EOL> self . closed = True <EOL> class _TelnetEventLoopInterface ( EventLoop ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , server ) : <EOL> self . _server = server <EOL> def close ( self ) : <EOL> "<STR_LIT>" <EOL> def stop ( self ) : <EOL> "<STR_LIT>" <EOL> def run_in_executor ( self , callback ) : <EOL> self . _server . run_in_executor ( callback ) <EOL> def call_from_executor ( self , callback , _max_postpone_until = None ) : <EOL> self . _server . call_from_executor ( callback ) <EOL> def add_reader ( self , fd , callback ) : <EOL> raise NotImplementedError <EOL> def remove_reader ( self , fd ) : <EOL> raise NotImplementedError <EOL> class TelnetServer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , host = '<STR_LIT:127.0.0.1>' , port = <NUM_LIT> , application = None , encoding = '<STR_LIT:utf-8>' ) : <EOL> assert isinstance ( host , text_type ) <EOL> assert isinstance ( port , int ) <EOL> assert isinstance ( application , TelnetApplication ) <EOL> assert isinstance ( encoding , text_type ) <EOL> self . host = host <EOL> self . port = port <EOL> self . application = application <EOL> self . encoding = encoding <EOL> self . connections = set ( ) <EOL> self . _calls_from_executor = [ ] <EOL> self . _schedule_pipe = os . pipe ( ) <EOL> fcntl . fcntl ( self . _schedule_pipe [ <NUM_LIT:0> ] , fcntl . F_SETFL , os . O_NONBLOCK ) <EOL> @ classmethod <EOL> def create_socket ( cls , host , port ) : <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> s . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> s . bind ( ( host , port ) ) <EOL> s . listen ( <NUM_LIT:4> ) <EOL> return s <EOL> def run_in_executor ( self , callback ) : <EOL> threading . Thread ( target = callback ) . start ( ) <EOL> def call_from_executor ( self , callback ) : <EOL> self . _calls_from_executor . append ( callback ) <EOL> if self . _schedule_pipe : <EOL> os . write ( self . _schedule_pipe [ <NUM_LIT:1> ] , b'<STR_LIT:x>' ) <EOL> def _process_callbacks ( self ) : <EOL> """<STR_LIT>""" <EOL> os . read ( self . _schedule_pipe [ <NUM_LIT:0> ] , <NUM_LIT> ) <EOL> calls_from_executor , self . _calls_from_executor = self . _calls_from_executor , [ ] <EOL> for c in calls_from_executor : <EOL> c ( ) <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> listen_socket = self . create_socket ( self . host , self . port ) <EOL> logger . info ( '<STR_LIT>' , self . host , self . port ) <EOL> try : <EOL> while True : <EOL> self . connections = set ( [ c for c in self . connections if not c . closed ] ) <EOL> connections = set ( [ c for c in self . connections if not c . handling_command ] ) <EOL> read_list = ( <EOL> [ listen_socket , self . _schedule_pipe [ <NUM_LIT:0> ] ] + <EOL> [ c . conn for c in connections ] ) <EOL> read , _ , _ = select . select ( read_list , [ ] , [ ] ) <EOL> for s in read : <EOL> if s == listen_socket : <EOL> self . _accept ( listen_socket ) <EOL> elif s == self . _schedule_pipe [ <NUM_LIT:0> ] : <EOL> self . _process_callbacks ( ) <EOL> else : <EOL> self . _handle_incoming_data ( s ) <EOL> finally : <EOL> listen_socket . close ( ) <EOL> def _accept ( self , listen_socket ) : <EOL> """<STR_LIT>""" <EOL> conn , addr = listen_socket . accept ( ) <EOL> connection = TelnetConnection ( conn , addr , self . application , self , encoding = self . encoding ) <EOL> self . connections . add ( connection ) <EOL> logger . info ( '<STR_LIT>' , * addr ) <EOL> def _handle_incoming_data ( self , conn ) : <EOL> """<STR_LIT>""" <EOL> connection = [ c for c in self . connections if c . conn == conn ] [ <NUM_LIT:0> ] <EOL> data = conn . recv ( <NUM_LIT> ) <EOL> if data : <EOL> connection . feed ( data ) <EOL> else : <EOL> self . connections . remove ( connection ) </s>
