<s> """<STR_LIT>""" <EOL> from __future__ import division , absolute_import <EOL> import tempfile <EOL> from zope . interface import implementer <EOL> from twisted . protocols import policies <EOL> from twisted . internet import interfaces , protocol , main , defer <EOL> from twisted . internet . task import deferLater <EOL> from twisted . python import failure <EOL> from twisted . internet . interfaces import IAddress <EOL> class _LoopbackQueue ( object ) : <EOL> """<STR_LIT>""" <EOL> _notificationDeferred = None <EOL> disconnect = False <EOL> def __init__ ( self ) : <EOL> self . _queue = [ ] <EOL> def put ( self , v ) : <EOL> self . _queue . append ( v ) <EOL> if self . _notificationDeferred is not None : <EOL> d , self . _notificationDeferred = self . _notificationDeferred , None <EOL> d . callback ( None ) <EOL> def __nonzero__ ( self ) : <EOL> return bool ( self . _queue ) <EOL> __bool__ = __nonzero__ <EOL> def get ( self ) : <EOL> return self . _queue . pop ( <NUM_LIT:0> ) <EOL> @ implementer ( IAddress ) <EOL> class _LoopbackAddress ( object ) : <EOL> pass <EOL> @ implementer ( interfaces . ITransport , interfaces . IConsumer ) <EOL> class _LoopbackTransport ( object ) : <EOL> disconnecting = False <EOL> producer = None <EOL> def __init__ ( self , q ) : <EOL> self . q = q <EOL> def write ( self , data ) : <EOL> if not isinstance ( data , bytes ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . q . put ( data ) <EOL> def writeSequence ( self , iovec ) : <EOL> self . q . put ( b'<STR_LIT>' . join ( iovec ) ) <EOL> def loseConnection ( self ) : <EOL> self . q . disconnect = True <EOL> self . q . put ( None ) <EOL> def abortConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . loseConnection ( ) <EOL> def getPeer ( self ) : <EOL> return _LoopbackAddress ( ) <EOL> def getHost ( self ) : <EOL> return _LoopbackAddress ( ) <EOL> def registerProducer ( self , producer , streaming ) : <EOL> assert self . producer is None <EOL> self . producer = producer <EOL> self . streamingProducer = streaming <EOL> self . _pollProducer ( ) <EOL> def unregisterProducer ( self ) : <EOL> assert self . producer is not None <EOL> self . producer = None <EOL> def _pollProducer ( self ) : <EOL> if self . producer is not None and not self . streamingProducer : <EOL> self . producer . resumeProducing ( ) <EOL> def identityPumpPolicy ( queue , target ) : <EOL> """<STR_LIT>""" <EOL> while queue : <EOL> bytes = queue . get ( ) <EOL> if bytes is None : <EOL> break <EOL> target . dataReceived ( bytes ) <EOL> def collapsingPumpPolicy ( queue , target ) : <EOL> """<STR_LIT>""" <EOL> bytes = [ ] <EOL> while queue : <EOL> chunk = queue . get ( ) <EOL> if chunk is None : <EOL> break <EOL> bytes . append ( chunk ) <EOL> if bytes : <EOL> target . dataReceived ( b'<STR_LIT>' . join ( bytes ) ) <EOL> def loopbackAsync ( server , client , pumpPolicy = identityPumpPolicy ) : <EOL> """<STR_LIT>""" <EOL> serverToClient = _LoopbackQueue ( ) <EOL> clientToServer = _LoopbackQueue ( ) <EOL> server . makeConnection ( _LoopbackTransport ( serverToClient ) ) <EOL> client . makeConnection ( _LoopbackTransport ( clientToServer ) ) <EOL> return _loopbackAsyncBody ( <EOL> server , serverToClient , client , clientToServer , pumpPolicy ) <EOL> def _loopbackAsyncBody ( server , serverToClient , client , clientToServer , <EOL> pumpPolicy ) : <EOL> """<STR_LIT>""" <EOL> def pump ( source , q , target ) : <EOL> sent = False <EOL> if q : <EOL> pumpPolicy ( q , target ) <EOL> sent = True <EOL> if sent and not q : <EOL> source . transport . _pollProducer ( ) <EOL> return sent <EOL> while <NUM_LIT:1> : <EOL> disconnect = clientSent = serverSent = False <EOL> serverSent = pump ( server , serverToClient , client ) <EOL> clientSent = pump ( client , clientToServer , server ) <EOL> if not clientSent and not serverSent : <EOL> d = defer . Deferred ( ) <EOL> clientToServer . _notificationDeferred = d <EOL> serverToClient . _notificationDeferred = d <EOL> d . addCallback ( <EOL> _loopbackAsyncContinue , <EOL> server , serverToClient , client , clientToServer , pumpPolicy ) <EOL> return d <EOL> if serverToClient . disconnect : <EOL> disconnect = True <EOL> pump ( server , serverToClient , client ) <EOL> elif clientToServer . disconnect : <EOL> disconnect = True <EOL> pump ( client , clientToServer , server ) <EOL> if disconnect : <EOL> server . connectionLost ( failure . Failure ( main . CONNECTION_DONE ) ) <EOL> client . connectionLost ( failure . Failure ( main . CONNECTION_DONE ) ) <EOL> return defer . succeed ( None ) <EOL> def _loopbackAsyncContinue ( ignored , server , serverToClient , client , <EOL> clientToServer , pumpPolicy ) : <EOL> clientToServer . _notificationDeferred = None <EOL> serverToClient . _notificationDeferred = None <EOL> from twisted . internet import reactor <EOL> return deferLater ( <EOL> reactor , <NUM_LIT:0> , <EOL> _loopbackAsyncBody , <EOL> server , serverToClient , client , clientToServer , pumpPolicy ) <EOL> @ implementer ( interfaces . ITransport , interfaces . IConsumer ) <EOL> class LoopbackRelay : <EOL> buffer = '<STR_LIT>' <EOL> shouldLose = <NUM_LIT:0> <EOL> disconnecting = <NUM_LIT:0> <EOL> producer = None <EOL> def __init__ ( self , target , logFile = None ) : <EOL> self . target = target <EOL> self . logFile = logFile <EOL> def write ( self , data ) : <EOL> self . buffer = self . buffer + data <EOL> if self . logFile : <EOL> self . logFile . write ( "<STR_LIT>" % repr ( data ) ) <EOL> def writeSequence ( self , iovec ) : <EOL> self . write ( "<STR_LIT>" . join ( iovec ) ) <EOL> def clearBuffer ( self ) : <EOL> if self . shouldLose == - <NUM_LIT:1> : <EOL> return <EOL> if self . producer : <EOL> self . producer . resumeProducing ( ) <EOL> if self . buffer : <EOL> if self . logFile : <EOL> self . logFile . write ( "<STR_LIT>" % repr ( self . buffer ) ) <EOL> buffer = self . buffer <EOL> self . buffer = '<STR_LIT>' <EOL> self . target . dataReceived ( buffer ) <EOL> if self . shouldLose == <NUM_LIT:1> : <EOL> self . shouldLose = - <NUM_LIT:1> <EOL> self . target . connectionLost ( failure . Failure ( main . CONNECTION_DONE ) ) <EOL> def loseConnection ( self ) : <EOL> if self . shouldLose != - <NUM_LIT:1> : <EOL> self . shouldLose = <NUM_LIT:1> <EOL> def getHost ( self ) : <EOL> return '<STR_LIT>' <EOL> def getPeer ( self ) : <EOL> return '<STR_LIT>' <EOL> def registerProducer ( self , producer , streaming ) : <EOL> self . producer = producer <EOL> def unregisterProducer ( self ) : <EOL> self . producer = None <EOL> def logPrefix ( self ) : <EOL> return '<STR_LIT>' % ( self . target . __class__ . __name__ , ) <EOL> class LoopbackClientFactory ( protocol . ClientFactory ) : <EOL> def __init__ ( self , protocol ) : <EOL> self . disconnected = <NUM_LIT:0> <EOL> self . deferred = defer . Deferred ( ) <EOL> self . protocol = protocol <EOL> def buildProtocol ( self , addr ) : <EOL> return self . protocol <EOL> def clientConnectionLost ( self , connector , reason ) : <EOL> self . disconnected = <NUM_LIT:1> <EOL> self . deferred . callback ( None ) <EOL> class _FireOnClose ( policies . ProtocolWrapper ) : <EOL> def __init__ ( self , protocol , factory ) : <EOL> policies . ProtocolWrapper . __init__ ( self , protocol , factory ) <EOL> self . deferred = defer . Deferred ( ) <EOL> def connectionLost ( self , reason ) : <EOL> policies . ProtocolWrapper . connectionLost ( self , reason ) <EOL> self . deferred . callback ( None ) <EOL> def loopbackTCP ( server , client , port = <NUM_LIT:0> , noisy = True ) : <EOL> """<STR_LIT>""" <EOL> from twisted . internet import reactor <EOL> f = policies . WrappingFactory ( protocol . Factory ( ) ) <EOL> serverWrapper = _FireOnClose ( f , server ) <EOL> f . noisy = noisy <EOL> f . buildProtocol = lambda addr : serverWrapper <EOL> serverPort = reactor . listenTCP ( port , f , interface = '<STR_LIT:127.0.0.1>' ) <EOL> clientF = LoopbackClientFactory ( client ) <EOL> clientF . noisy = noisy <EOL> reactor . connectTCP ( '<STR_LIT:127.0.0.1>' , serverPort . getHost ( ) . port , clientF ) <EOL> d = clientF . deferred <EOL> d . addCallback ( lambda x : serverWrapper . deferred ) <EOL> d . addCallback ( lambda x : serverPort . stopListening ( ) ) <EOL> return d <EOL> def loopbackUNIX ( server , client , noisy = True ) : <EOL> """<STR_LIT>""" <EOL> path = tempfile . mktemp ( ) <EOL> from twisted . internet import reactor <EOL> f = policies . WrappingFactory ( protocol . Factory ( ) ) <EOL> serverWrapper = _FireOnClose ( f , server ) <EOL> f . noisy = noisy <EOL> f . buildProtocol = lambda addr : serverWrapper <EOL> serverPort = reactor . listenUNIX ( path , f ) <EOL> clientF = LoopbackClientFactory ( client ) <EOL> clientF . noisy = noisy <EOL> reactor . connectUNIX ( path , clientF ) <EOL> d = clientF . deferred <EOL> d . addCallback ( lambda x : serverWrapper . deferred ) <EOL> d . addCallback ( lambda x : serverPort . stopListening ( ) ) <EOL> return d </s>
