#NEWLINE# def _maybe_connect( self , to_pid , callback = None ) : #NEWLINE# #INDENT# 'Asynchronously establish a connection to the remote pid.' #NEWLINE# callback = stack_context . wrap ( ( callback or ( lambda stream : None ) ) ) #NEWLINE# def streaming_callback( data ) : #NEWLINE# #INDENT# log . info ( ( 'Received %d bytes from %s, discarding.' % ( len ( data ) , to_pid ) ) ) #NEWLINE# log . debug ( ( '  data: %r' % ( data , ) ) ) #NEWLINE# #UNINDENT# def on_connect( exit_cb , stream ) : #NEWLINE# #INDENT# log . info ( ( 'Connection to %s established' % to_pid ) ) #NEWLINE# with self . _connection_callbacks_lock : #NEWLINE# #INDENT# self . _connections [ to_pid ] = stream #NEWLINE# #UNINDENT# self . __dispatch_on_connect_callbacks ( to_pid , stream ) #NEWLINE# self . __loop . add_callback ( stream . read_until_close , exit_cb , streaming_callback = streaming_callback ) #NEWLINE# #UNINDENT# create = False #NEWLINE# with self . _connection_callbacks_lock : #NEWLINE# #INDENT# stream = self . _connections . get ( to_pid ) #NEWLINE# callbacks = self . _connection_callbacks . get ( to_pid ) #NEWLINE# if ( not stream ) : #NEWLINE# #INDENT# self . _connection_callbacks [ to_pid ] . append ( callback ) #NEWLINE# if ( not callbacks ) : #NEWLINE# #INDENT# create = True #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# if stream : #NEWLINE# #INDENT# self . __loop . add_callback ( callback , stream ) #NEWLINE# return #NEWLINE# #UNINDENT# if ( not stream ) : #NEWLINE# #INDENT# return #NEWLINE# #UNINDENT# sock = socket . socket ( socket . AF_INET , socket . SOCK_STREAM , 0 ) #NEWLINE# if ( not sock ) : #NEWLINE# #INDENT# raise self . SocketError ( 'Failed opening socket' ) #NEWLINE# #UNINDENT# stream = IOStream ( sock , io_loop = self . __loop ) #NEWLINE# stream . set_nodelay ( True ) #NEWLINE# stream . set_close_callback ( partial ( self . __on_exit , to_pid , b'reached end of stream' ) ) #NEWLINE# connect_callback = partial ( on_connect , partial ( self . __on_exit , to_pid ) , stream ) #NEWLINE# log . info ( ( 'Establishing connection to %s' % to_pid ) ) #NEWLINE# stream . connect ( ( to_pid . ip , to_pid . port ) , callback = connect_callback ) #NEWLINE# if stream . closed ( ) : #NEWLINE# #INDENT# raise self . SocketError ( 'Failed to initiate stream connection' ) #NEWLINE# #UNINDENT# log . info ( ( 'Maybe connected to %s' % to_pid ) )