<s> import threading <EOL> import ufora . networking . Transport as Transport <EOL> import ufora . util . ManagedThread as ManagedThread <EOL> import ufora . util . Deferred as Deferred <EOL> class InMemoryTransport ( Transport . Transport ) : <EOL> ConnectMessage = '<STR_LIT>' <EOL> LogMessage = '<STR_LIT>' <EOL> DisconnectMessage = '<STR_LIT>' <EOL> def __init__ ( self , channel ) : <EOL> self . channel = channel <EOL> self . onMessageReceived = None <EOL> self . onDisconnected = None <EOL> self . inputThread = None <EOL> self . isShuttingDown = False <EOL> def connect ( self , credentials ) : <EOL> assert self . onMessageReceived is not None , "<STR_LIT>" <EOL> assert self . onDisconnected is not None , "<STR_LIT>" <EOL> assert not self . isInInputThread ( ) , "<STR_LIT>" <EOL> deferred = Deferred . Deferred ( ) <EOL> self . startInputLoop ( deferred ) <EOL> if credentials is not None : <EOL> self . channel . write ( '<STR_LIT>' % credentials ) <EOL> return deferred <EOL> def startInputLoop ( self , deferred ) : <EOL> assert self . inputThread is None or not self . inputThread . is_alive ( ) <EOL> self . inputThread = ManagedThread . ManagedThread ( target = self . inputLoop , args = ( deferred , ) ) <EOL> self . inputThread . start ( ) <EOL> def send ( self , content ) : <EOL> self . channel . write ( content ) <EOL> def disconnect ( self ) : <EOL> self . isShuttingDown = True <EOL> self . channel . write ( InMemoryTransport . DisconnectMessage ) <EOL> if not self . isInInputThread ( ) : <EOL> self . blockUntilFullyDisconnected ( ) <EOL> def blockUntilFullyDisconnected ( self ) : <EOL> self . inputThread . join ( ) <EOL> def isInInputThread ( self ) : <EOL> return self . inputThread is not None and threading . currentThread ( ) . ident == self . inputThread . ident <EOL> def inputLoop ( self , connectDeferred ) : <EOL> try : <EOL> if connectDeferred and not self . waitForConnection ( connectDeferred ) : <EOL> return <EOL> self . processIncomingMessages ( ) <EOL> finally : <EOL> self . isShuttingDown = False <EOL> def processIncomingMessages ( self ) : <EOL> while not self . isShuttingDown : <EOL> message = self . channel . get ( ) <EOL> if message == InMemoryTransport . DisconnectMessage : <EOL> self . onDisconnected ( ) <EOL> return <EOL> try : <EOL> self . onMessageReceived ( message ) <EOL> except TypeError as e : <EOL> print '<STR_LIT>' % ( e , message ) <EOL> def waitForConnection ( self , connectDeferred ) : <EOL> message = None <EOL> while True : <EOL> message = self . channel . get ( ) <EOL> if message . startswith ( InMemoryTransport . ConnectMessage ) : <EOL> break <EOL> result = message [ len ( InMemoryTransport . ConnectMessage ) : ] . split ( '<STR_LIT:U+002C>' ) <EOL> if result [ <NUM_LIT:0> ] == '<STR_LIT:OK>' : <EOL> connectDeferred . callback ( { <EOL> '<STR_LIT>' : result [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : result [ <NUM_LIT:2> ] , <EOL> '<STR_LIT>' : result [ <NUM_LIT:3> ] , <EOL> '<STR_LIT>' : result [ <NUM_LIT:4> ] <EOL> } ) <EOL> return True <EOL> else : <EOL> connectDeferred . errback ( '<STR_LIT>' ) <EOL> return False </s>
