<s> import logging <EOL> import collections <EOL> import re <EOL> from tornado import simple_httpclient <EOL> from tornado . ioloop import IOLoop <EOL> from tornado import httpclient , httputil <EOL> from tornado . concurrent import TracebackFuture <EOL> try : <EOL> from tornado . tcpclient import TCPClient <EOL> old_tornado = False <EOL> except ImportError : <EOL> old_tornado = True <EOL> from tornado . netutil import Resolver <EOL> from tornado . escape import native_str <EOL> class EventSourceError ( Exception ) : <EOL> pass <EOL> class HeadersData ( object ) : <EOL> def __init__ ( self , code , reason ) : <EOL> self . code = code <EOL> self . reason = reason <EOL> class Event ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . name = None <EOL> self . data = None <EOL> self . id = None <EOL> self . retry = None <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( str ( self . id ) , str ( self . name ) , str ( self . data . replace ( "<STR_LIT:\n>" , "<STR_LIT>" ) ) ) <EOL> class EventSourceClient ( simple_httpclient . _HTTPConnection ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , io_loop , request ) : <EOL> self . connect_future = TracebackFuture ( ) <EOL> self . read_future = None <EOL> self . read_queue = collections . deque ( ) <EOL> self . events = [ ] <EOL> if old_tornado : <EOL> self . resolver = Resolver ( io_loop = io_loop ) <EOL> super ( EventSourceClient , self ) . __init__ ( <EOL> io_loop , None , request , lambda : None , self . _on_http_response , <EOL> <NUM_LIT> , self . resolver ) <EOL> else : <EOL> self . tcp_client = TCPClient ( io_loop = io_loop ) <EOL> super ( EventSourceClient , self ) . __init__ ( <EOL> io_loop , None , request , lambda : None , self . _on_http_response , <EOL> <NUM_LIT> , self . tcp_client , <NUM_LIT> ) <EOL> def _handle_event_stream ( self ) : <EOL> if self . _timeout is not None : <EOL> self . io_loop . remove_timeout ( self . _timeout ) <EOL> self . _timeout = None <EOL> self . stream . read_until_regex ( b"<STR_LIT>" , self . handle_stream ) <EOL> self . connect_future . set_result ( self ) <EOL> def _on_http_response ( self , response ) : <EOL> if not self . connect_future . done ( ) : <EOL> if response . error : <EOL> self . connect_future . set_exception ( response . error ) <EOL> else : <EOL> self . connect_future . set_exception ( EventSourceError ( <EOL> "<STR_LIT>" ) ) <EOL> def _on_headers ( self , data ) : <EOL> data = native_str ( data . decode ( "<STR_LIT>" ) ) <EOL> first_line , _ , header_data = data . partition ( "<STR_LIT:\n>" ) <EOL> match = re . match ( "<STR_LIT>" , first_line ) <EOL> assert match <EOL> code = int ( match . group ( <NUM_LIT:1> ) ) <EOL> headers = httputil . HTTPHeaders . parse ( header_data ) <EOL> reason = match . group ( <NUM_LIT:2> ) <EOL> self . headers_received ( HeadersData ( code = code , reason = reason ) , headers ) <EOL> def headers_received ( self , data , headers ) : <EOL> self . headers = headers <EOL> self . code = data . code <EOL> self . reason = data . reason <EOL> if self . headers [ '<STR_LIT:Content-Type>' ] != '<STR_LIT>' : <EOL> raise ValueError ( "<STR_LIT>" % self . headers ) <EOL> if "<STR_LIT>" in self . headers : <EOL> if "<STR_LIT:U+002C>" in self . headers [ "<STR_LIT>" ] : <EOL> pieces = re . split ( r'<STR_LIT>' , self . headers [ "<STR_LIT>" ] ) <EOL> if any ( i != pieces [ <NUM_LIT:0> ] for i in pieces ) : <EOL> raise ValueError ( "<STR_LIT>" % <EOL> self . headers [ "<STR_LIT>" ] ) <EOL> self . headers [ "<STR_LIT>" ] = pieces [ <NUM_LIT:0> ] <EOL> self . _handle_event_stream ( ) <EOL> def handle_stream ( self , message ) : <EOL> """<STR_LIT>""" <EOL> logging . debug ( "<STR_LIT>" ) <EOL> event = Event ( ) <EOL> for line in message . strip ( ) . splitlines ( ) : <EOL> ( field , value ) = line . split ( "<STR_LIT::>" , <NUM_LIT:1> ) <EOL> field = field . strip ( ) <EOL> if field == "<STR_LIT>" : <EOL> event . name = value . lstrip ( ) <EOL> elif field == "<STR_LIT:data>" : <EOL> value = value . lstrip ( ) <EOL> if event . data is None : <EOL> event . data = value <EOL> else : <EOL> event . data = "<STR_LIT>" % ( event . data , value ) <EOL> elif field == "<STR_LIT:id>" : <EOL> event . id = value . lstrip ( ) <EOL> self . last_event_id = event . id <EOL> elif field == "<STR_LIT>" : <EOL> try : <EOL> self . retry_timeout = int ( value ) <EOL> event . retry = self . retry_timeout <EOL> logging . info ( "<STR_LIT>" % ( value , ) ) <EOL> except ValueError : <EOL> pass <EOL> elif field == "<STR_LIT>" : <EOL> logging . debug ( "<STR_LIT>" % ( value , ) ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> self . events . append ( event ) <EOL> def eventsource_connect ( url , io_loop = None , callback = None , connect_timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if io_loop is None : <EOL> io_loop = IOLoop . current ( ) <EOL> if isinstance ( url , httpclient . HTTPRequest ) : <EOL> assert connect_timeout is None <EOL> request = url <EOL> request . headers = httputil . HTTPHeaders ( request . headers ) <EOL> else : <EOL> request = httpclient . HTTPRequest ( <EOL> url , <EOL> connect_timeout = connect_timeout , <EOL> headers = httputil . HTTPHeaders ( { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } ) <EOL> ) <EOL> request = httpclient . _RequestProxy ( <EOL> request , httpclient . HTTPRequest . _DEFAULTS ) <EOL> conn = EventSourceClient ( io_loop , request ) <EOL> if callback is not None : <EOL> io_loop . add_future ( conn . connect_future , callback ) <EOL> return conn . connect_future </s>
